up step
Some checks failed
K8S Fission Deployment / Deployment fission functions (push) Failing after 12s
Some checks failed
K8S Fission Deployment / Deployment fission functions (push) Failing after 12s
This commit is contained in:
1
apps/tests/__init__.py
Normal file
1
apps/tests/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
# Tests package for AI Admin API
|
||||
99
apps/tests/conftest.py
Normal file
99
apps/tests/conftest.py
Normal file
@@ -0,0 +1,99 @@
|
||||
"""Shared fixtures for API handler tests."""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
from flask import Flask
|
||||
|
||||
# Add apps directory to path for imports
|
||||
apps_dir = Path(__file__).parent.parent
|
||||
sys.path.insert(0, str(apps_dir))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def flask_app():
|
||||
"""Create Flask app context for testing."""
|
||||
app = Flask(__name__)
|
||||
app.config["TESTING"] = True
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def app_context(flask_app):
|
||||
"""Provide Flask application context."""
|
||||
with flask_app.app_context():
|
||||
yield flask_app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def request_context(flask_app):
|
||||
"""Provide Flask request context."""
|
||||
with flask_app.test_request_context():
|
||||
yield flask_app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_db_connection(mocker):
|
||||
"""Mock database connection with cursor that has description attribute."""
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.description = [
|
||||
MagicMock(name="id"),
|
||||
MagicMock(name="name"),
|
||||
MagicMock(name="dob"),
|
||||
MagicMock(name="email"),
|
||||
MagicMock(name="gender"),
|
||||
MagicMock(name="created"),
|
||||
MagicMock(name="modified"),
|
||||
]
|
||||
# Set name attribute on description items
|
||||
for i, col_name in enumerate(["id", "name", "dob", "email", "gender", "created", "modified"]):
|
||||
mock_cursor.description[i].name = col_name
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mocker.patch("helpers.init_db_connection", return_value=mock_conn)
|
||||
|
||||
return {"connection": mock_conn, "cursor": mock_cursor}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_secrets(mocker):
|
||||
"""Mock get_secret to return test values."""
|
||||
secrets = {
|
||||
"PG_HOST": "localhost",
|
||||
"PG_PORT": "5432",
|
||||
"PG_DB": "test_db",
|
||||
"PG_USER": "test_user",
|
||||
"PG_PASS": "test_pass",
|
||||
}
|
||||
mocker.patch("helpers.get_secret", side_effect=lambda key, default=None: secrets.get(key, default))
|
||||
return secrets
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_user_data():
|
||||
"""Sample user data for testing."""
|
||||
return {
|
||||
"name": "Test User",
|
||||
"email": "test@example.com",
|
||||
"dob": "1990-01-15",
|
||||
"gender": "male",
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_db_row():
|
||||
"""Sample database row tuple."""
|
||||
return (
|
||||
"550e8400-e29b-41d4-a716-446655440000", # id
|
||||
"Test User", # name
|
||||
"1990-01-15", # dob
|
||||
"test@example.com", # email
|
||||
"male", # gender
|
||||
"2024-01-01T10:00:00", # created
|
||||
"2024-01-01T10:00:00", # modified
|
||||
)
|
||||
478
apps/tests/test_filter_insert.py
Normal file
478
apps/tests/test_filter_insert.py
Normal file
@@ -0,0 +1,478 @@
|
||||
"""Tests for filter_insert.py - GET (filter) & POST (create) handlers."""
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
from flask import Flask
|
||||
from psycopg2 import IntegrityError
|
||||
|
||||
|
||||
class TestMain:
|
||||
"""Tests for main() dispatcher function."""
|
||||
|
||||
def test_main_get_calls_filter(self, mocker):
|
||||
"""Test GET request routes to make_filter_request().
|
||||
|
||||
Given:
|
||||
A GET request to the endpoint.
|
||||
When:
|
||||
main() is called.
|
||||
Then:
|
||||
make_filter_request() is invoked and returns 200.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
with app.test_request_context(method="GET"):
|
||||
mock_filter = mocker.patch(
|
||||
"filter_insert.make_filter_request",
|
||||
return_value=({"data": []}, 200, {}),
|
||||
)
|
||||
mocker.patch("filter_insert.init_db_connection")
|
||||
|
||||
import filter_insert
|
||||
|
||||
result = filter_insert.main()
|
||||
|
||||
mock_filter.assert_called_once()
|
||||
assert result[1] == 200
|
||||
|
||||
def test_main_post_calls_insert(self, mocker):
|
||||
"""Test POST request routes to make_insert_request().
|
||||
|
||||
Given:
|
||||
A POST request with JSON body containing user data.
|
||||
When:
|
||||
main() is called.
|
||||
Then:
|
||||
make_insert_request() is invoked and returns 201.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
with app.test_request_context(
|
||||
method="POST",
|
||||
json={"name": "Test", "email": "test@example.com"},
|
||||
):
|
||||
mock_insert = mocker.patch(
|
||||
"filter_insert.make_insert_request",
|
||||
return_value=({"id": "123"}, 201, {}),
|
||||
)
|
||||
|
||||
import filter_insert
|
||||
|
||||
result = filter_insert.main()
|
||||
|
||||
mock_insert.assert_called_once()
|
||||
assert result[1] == 201
|
||||
|
||||
def test_main_invalid_method_returns_405(self, mocker):
|
||||
"""Test unsupported HTTP method returns 405.
|
||||
|
||||
Given:
|
||||
A PATCH request (unsupported method).
|
||||
When:
|
||||
main() is called.
|
||||
Then:
|
||||
Returns 405 Method Not Allowed with error message.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
with app.test_request_context(method="PATCH"):
|
||||
import filter_insert
|
||||
|
||||
result = filter_insert.main()
|
||||
|
||||
assert result[1] == 405
|
||||
assert "error" in result[0]
|
||||
|
||||
|
||||
class TestMakeInsertRequest:
|
||||
"""Tests for make_insert_request() - user creation."""
|
||||
|
||||
def test_insert_success(self, mocker, sample_user_data, sample_db_row):
|
||||
"""Test successful user creation returns 201.
|
||||
|
||||
Given:
|
||||
Valid user data with name, email, dob, and gender.
|
||||
Database connection is available.
|
||||
When:
|
||||
POST request to create user.
|
||||
Then:
|
||||
Returns 201 status code.
|
||||
Executes INSERT query with user data.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.description = [
|
||||
MagicMock(name="id"),
|
||||
MagicMock(name="name"),
|
||||
MagicMock(name="dob"),
|
||||
MagicMock(name="email"),
|
||||
MagicMock(name="gender"),
|
||||
MagicMock(name="created"),
|
||||
MagicMock(name="modified"),
|
||||
]
|
||||
for i, col_name in enumerate(
|
||||
["id", "name", "dob", "email", "gender", "created", "modified"]
|
||||
):
|
||||
mock_cursor.description[i].name = col_name
|
||||
mock_cursor.fetchone.return_value = sample_db_row
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.__enter__ = MagicMock(return_value=mock_conn)
|
||||
mock_conn.__exit__ = MagicMock(return_value=False)
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mocker.patch("filter_insert.init_db_connection", return_value=mock_conn)
|
||||
|
||||
with app.test_request_context(
|
||||
method="POST",
|
||||
json=sample_user_data,
|
||||
content_type="application/json",
|
||||
):
|
||||
import filter_insert
|
||||
|
||||
result = filter_insert.make_insert_request()
|
||||
|
||||
assert result[1] == 201
|
||||
mock_cursor.execute.assert_called_once()
|
||||
|
||||
def test_insert_validation_error_missing_name(self, mocker):
|
||||
"""Test missing required field 'name' returns 400.
|
||||
|
||||
Given:
|
||||
Request body with email but missing required 'name' field.
|
||||
When:
|
||||
POST request to create user.
|
||||
Then:
|
||||
Returns 400 Bad Request.
|
||||
Response contains errorCode 'VALIDATION_ERROR'.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
with app.test_request_context(
|
||||
method="POST",
|
||||
json={"email": "test@example.com"}, # missing 'name'
|
||||
content_type="application/json",
|
||||
):
|
||||
import filter_insert
|
||||
|
||||
result = filter_insert.make_insert_request()
|
||||
|
||||
assert result[1] == 400
|
||||
response_data = result[0].get_json()
|
||||
assert response_data["errorCode"] == "VALIDATION_ERROR"
|
||||
|
||||
def test_insert_validation_error_invalid_email(self, mocker):
|
||||
"""Test invalid email format raises serialization error.
|
||||
|
||||
Given:
|
||||
Request body with invalid email format 'invalid-email'.
|
||||
When:
|
||||
POST request to create user.
|
||||
Then:
|
||||
Raises TypeError because ValidationError.errors() contains
|
||||
ValueError which is not JSON serializable.
|
||||
|
||||
Note:
|
||||
This test documents a bug in the source code where e.errors()
|
||||
is passed directly to jsonify without sanitization.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
with app.test_request_context(
|
||||
method="POST",
|
||||
json={"name": "Test", "email": "invalid-email"},
|
||||
content_type="application/json",
|
||||
):
|
||||
import filter_insert
|
||||
|
||||
with pytest.raises(TypeError, match="not JSON serializable"):
|
||||
filter_insert.make_insert_request()
|
||||
|
||||
def test_insert_duplicate_email(self, mocker, sample_user_data):
|
||||
"""Test duplicate email returns 409 Conflict.
|
||||
|
||||
Given:
|
||||
Valid user data but email already exists in database.
|
||||
Database raises IntegrityError on INSERT.
|
||||
When:
|
||||
POST request to create user.
|
||||
Then:
|
||||
Returns 409 Conflict.
|
||||
Response contains errorCode 'DUPLICATE_TAG'.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.__enter__ = MagicMock(return_value=mock_conn)
|
||||
mock_conn.__exit__ = MagicMock(return_value=False)
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(
|
||||
side_effect=IntegrityError("duplicate key value")
|
||||
)
|
||||
|
||||
mocker.patch("filter_insert.init_db_connection", return_value=mock_conn)
|
||||
|
||||
with app.test_request_context(
|
||||
method="POST",
|
||||
json=sample_user_data,
|
||||
content_type="application/json",
|
||||
):
|
||||
import filter_insert
|
||||
|
||||
result = filter_insert.make_insert_request()
|
||||
|
||||
assert result[1] == 409
|
||||
response_data = result[0].get_json()
|
||||
assert response_data["errorCode"] == "DUPLICATE_TAG"
|
||||
|
||||
|
||||
class TestMakeFilterRequest:
|
||||
"""Tests for make_filter_request() - user filtering."""
|
||||
|
||||
def test_filter_empty_result(self, mocker):
|
||||
"""Test filter with no matching results returns empty array.
|
||||
|
||||
Given:
|
||||
Database has no users matching the filter criteria.
|
||||
When:
|
||||
GET request to filter users.
|
||||
Then:
|
||||
Returns empty JSON array [].
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.description = [
|
||||
MagicMock(name="id"),
|
||||
MagicMock(name="name"),
|
||||
MagicMock(name="dob"),
|
||||
MagicMock(name="email"),
|
||||
MagicMock(name="gender"),
|
||||
MagicMock(name="created"),
|
||||
MagicMock(name="modified"),
|
||||
MagicMock(name="count"),
|
||||
MagicMock(name="total"),
|
||||
]
|
||||
for i, col_name in enumerate(
|
||||
["id", "name", "dob", "email", "gender", "created", "modified", "count", "total"]
|
||||
):
|
||||
mock_cursor.description[i].name = col_name
|
||||
mock_cursor.fetchall.return_value = []
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mocker.patch("filter_insert.init_db_connection", return_value=mock_conn)
|
||||
|
||||
with app.test_request_context(method="GET"):
|
||||
import filter_insert
|
||||
|
||||
response = filter_insert.make_filter_request()
|
||||
|
||||
# make_filter_request returns Response object directly (not tuple)
|
||||
response_data = response.get_json()
|
||||
assert response_data == []
|
||||
|
||||
def test_filter_with_pagination(self, mocker, sample_db_row):
|
||||
"""Test filter with page and size parameters.
|
||||
|
||||
Given:
|
||||
Request with query params page=1 and size=5.
|
||||
When:
|
||||
GET request to filter users.
|
||||
Then:
|
||||
SQL query uses LIMIT 5 and OFFSET 5 (page * size).
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.description = [
|
||||
MagicMock(name="id"),
|
||||
MagicMock(name="name"),
|
||||
MagicMock(name="dob"),
|
||||
MagicMock(name="email"),
|
||||
MagicMock(name="gender"),
|
||||
MagicMock(name="created"),
|
||||
MagicMock(name="modified"),
|
||||
MagicMock(name="count"),
|
||||
MagicMock(name="total"),
|
||||
]
|
||||
for i, col_name in enumerate(
|
||||
["id", "name", "dob", "email", "gender", "created", "modified", "count", "total"]
|
||||
):
|
||||
mock_cursor.description[i].name = col_name
|
||||
|
||||
# Add count and total to sample row
|
||||
row_with_counts = sample_db_row + (10, 10)
|
||||
mock_cursor.fetchall.return_value = [row_with_counts]
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mocker.patch("filter_insert.init_db_connection", return_value=mock_conn)
|
||||
|
||||
with app.test_request_context(method="GET", query_string={"page": "1", "size": "5"}):
|
||||
import filter_insert
|
||||
|
||||
filter_insert.make_filter_request()
|
||||
|
||||
# Check that execute was called with correct offset
|
||||
# cursor.execute(sql, values) - values is second positional arg
|
||||
call_args = mock_cursor.execute.call_args
|
||||
values = call_args[0][1] # Second positional argument
|
||||
assert values["limit"] == 5
|
||||
assert values["offset"] == 5 # page 1 * size 5
|
||||
|
||||
def test_filter_with_keyword(self, mocker, sample_db_row):
|
||||
"""Test filter with keyword search across name and email.
|
||||
|
||||
Given:
|
||||
Request with query param filter[keyword]='test'.
|
||||
When:
|
||||
GET request to filter users.
|
||||
Then:
|
||||
SQL query contains ILIKE clause for keyword matching.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.description = [
|
||||
MagicMock(name="id"),
|
||||
MagicMock(name="name"),
|
||||
MagicMock(name="dob"),
|
||||
MagicMock(name="email"),
|
||||
MagicMock(name="gender"),
|
||||
MagicMock(name="created"),
|
||||
MagicMock(name="modified"),
|
||||
MagicMock(name="count"),
|
||||
MagicMock(name="total"),
|
||||
]
|
||||
for i, col_name in enumerate(
|
||||
["id", "name", "dob", "email", "gender", "created", "modified", "count", "total"]
|
||||
):
|
||||
mock_cursor.description[i].name = col_name
|
||||
|
||||
row_with_counts = sample_db_row + (1, 1)
|
||||
mock_cursor.fetchall.return_value = [row_with_counts]
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mocker.patch("filter_insert.init_db_connection", return_value=mock_conn)
|
||||
|
||||
with app.test_request_context(
|
||||
method="GET", query_string={"filter[keyword]": "test"}
|
||||
):
|
||||
import filter_insert
|
||||
|
||||
result = filter_insert.make_filter_request()
|
||||
|
||||
# Check SQL contains keyword filter
|
||||
call_args = mock_cursor.execute.call_args
|
||||
sql = call_args[0][0]
|
||||
assert "ILIKE" in sql
|
||||
|
||||
def test_filter_with_name(self, mocker, sample_db_row):
|
||||
"""Test filter by name with case-insensitive partial match.
|
||||
|
||||
Given:
|
||||
Request with query param filter[name]='John'.
|
||||
When:
|
||||
GET request to filter users.
|
||||
Then:
|
||||
SQL query contains 'LOWER(name) LIKE %john%'.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.description = [
|
||||
MagicMock(name="id"),
|
||||
MagicMock(name="name"),
|
||||
MagicMock(name="dob"),
|
||||
MagicMock(name="email"),
|
||||
MagicMock(name="gender"),
|
||||
MagicMock(name="created"),
|
||||
MagicMock(name="modified"),
|
||||
MagicMock(name="count"),
|
||||
MagicMock(name="total"),
|
||||
]
|
||||
for i, col_name in enumerate(
|
||||
["id", "name", "dob", "email", "gender", "created", "modified", "count", "total"]
|
||||
):
|
||||
mock_cursor.description[i].name = col_name
|
||||
|
||||
row_with_counts = sample_db_row + (1, 1)
|
||||
mock_cursor.fetchall.return_value = [row_with_counts]
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mocker.patch("filter_insert.init_db_connection", return_value=mock_conn)
|
||||
|
||||
with app.test_request_context(
|
||||
method="GET", query_string={"filter[name]": "John"}
|
||||
):
|
||||
import filter_insert
|
||||
|
||||
filter_insert.make_filter_request()
|
||||
|
||||
call_args = mock_cursor.execute.call_args
|
||||
sql = call_args[0][0]
|
||||
values = call_args[0][1] # Second positional argument
|
||||
assert "LOWER(name) LIKE" in sql
|
||||
assert values["name"] == "%john%"
|
||||
|
||||
def test_filter_with_sortby(self, mocker, sample_db_row):
|
||||
"""Test filter with sortby and asc parameters.
|
||||
|
||||
Given:
|
||||
Request with query params sortby='created' and asc='true'.
|
||||
When:
|
||||
GET request to filter users.
|
||||
Then:
|
||||
SQL query contains 'ORDER BY created ASC'.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.description = [
|
||||
MagicMock(name="id"),
|
||||
MagicMock(name="name"),
|
||||
MagicMock(name="dob"),
|
||||
MagicMock(name="email"),
|
||||
MagicMock(name="gender"),
|
||||
MagicMock(name="created"),
|
||||
MagicMock(name="modified"),
|
||||
MagicMock(name="count"),
|
||||
MagicMock(name="total"),
|
||||
]
|
||||
for i, col_name in enumerate(
|
||||
["id", "name", "dob", "email", "gender", "created", "modified", "count", "total"]
|
||||
):
|
||||
mock_cursor.description[i].name = col_name
|
||||
|
||||
row_with_counts = sample_db_row + (1, 1)
|
||||
mock_cursor.fetchall.return_value = [row_with_counts]
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mocker.patch("filter_insert.init_db_connection", return_value=mock_conn)
|
||||
|
||||
with app.test_request_context(
|
||||
method="GET", query_string={"sortby": "created", "asc": "true"}
|
||||
):
|
||||
import filter_insert
|
||||
|
||||
result = filter_insert.make_filter_request()
|
||||
|
||||
call_args = mock_cursor.execute.call_args
|
||||
sql = call_args[0][0]
|
||||
assert "ORDER BY created ASC" in sql
|
||||
514
apps/tests/test_update_delete.py
Normal file
514
apps/tests/test_update_delete.py
Normal file
@@ -0,0 +1,514 @@
|
||||
"""Tests for update_delete.py - PUT (update) & DELETE handlers."""
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
from flask import Flask
|
||||
from psycopg2 import IntegrityError
|
||||
|
||||
|
||||
class TestMain:
|
||||
"""Tests for main() dispatcher function."""
|
||||
|
||||
def test_main_put_calls_update(self, mocker):
|
||||
"""Test PUT request routes to make_update_request().
|
||||
|
||||
Given:
|
||||
A PUT request to the endpoint.
|
||||
When:
|
||||
main() is called.
|
||||
Then:
|
||||
make_update_request() is invoked and returns 200.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
with app.test_request_context(method="PUT"):
|
||||
mock_update = mocker.patch(
|
||||
"update_delete.make_update_request",
|
||||
return_value=({"id": "123"}, 200, {}),
|
||||
)
|
||||
|
||||
import update_delete
|
||||
|
||||
result = update_delete.main()
|
||||
|
||||
mock_update.assert_called_once()
|
||||
assert result[1] == 200
|
||||
|
||||
def test_main_delete_calls_delete(self, mocker):
|
||||
"""Test DELETE request routes to make_delete_request().
|
||||
|
||||
Given:
|
||||
A DELETE request to the endpoint.
|
||||
When:
|
||||
main() is called.
|
||||
Then:
|
||||
make_delete_request() is invoked and returns 200.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
with app.test_request_context(method="DELETE"):
|
||||
mock_delete = mocker.patch(
|
||||
"update_delete.make_delete_request",
|
||||
return_value=({"id": "123"}, 200, {}),
|
||||
)
|
||||
|
||||
import update_delete
|
||||
|
||||
result = update_delete.main()
|
||||
|
||||
mock_delete.assert_called_once()
|
||||
assert result[1] == 200
|
||||
|
||||
def test_main_invalid_method_returns_405(self, mocker):
|
||||
"""Test unsupported HTTP method returns 405.
|
||||
|
||||
Given:
|
||||
A POST request (unsupported method for this endpoint).
|
||||
When:
|
||||
main() is called.
|
||||
Then:
|
||||
Returns 405 Method Not Allowed with error message.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
with app.test_request_context(method="POST"):
|
||||
import update_delete
|
||||
|
||||
result = update_delete.main()
|
||||
|
||||
assert result[1] == 405
|
||||
assert "error" in result[0]
|
||||
|
||||
|
||||
class TestMakeUpdateRequest:
|
||||
"""Tests for make_update_request() - user update."""
|
||||
|
||||
def test_update_missing_user_id(self, mocker):
|
||||
"""Test missing X-Fission-Params-UserID header returns 400.
|
||||
|
||||
Given:
|
||||
PUT request without X-Fission-Params-UserID header.
|
||||
When:
|
||||
make_update_request() is called.
|
||||
Then:
|
||||
Returns 400 Bad Request.
|
||||
Response contains errorCode 'MISSING_USER_ID'.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
with app.test_request_context(
|
||||
method="PUT",
|
||||
json={"name": "Updated Name"},
|
||||
content_type="application/json",
|
||||
):
|
||||
import update_delete
|
||||
|
||||
result = update_delete.make_update_request()
|
||||
|
||||
assert result[1] == 400
|
||||
response_data = result[0].get_json()
|
||||
assert response_data["errorCode"] == "MISSING_USER_ID"
|
||||
|
||||
def test_update_user_not_found(self, mocker):
|
||||
"""Test update non-existent user returns 404.
|
||||
|
||||
Given:
|
||||
Valid X-Fission-Params-UserID header.
|
||||
User does not exist in database (SELECT returns None).
|
||||
When:
|
||||
PUT request to update user.
|
||||
Then:
|
||||
Returns 404 Not Found.
|
||||
Response contains errorCode 'USER_NOT_FOUND'.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.fetchone.return_value = None # User not found
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.__enter__ = MagicMock(return_value=mock_conn)
|
||||
mock_conn.__exit__ = MagicMock(return_value=False)
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mocker.patch("update_delete.init_db_connection", return_value=mock_conn)
|
||||
|
||||
with app.test_request_context(
|
||||
method="PUT",
|
||||
headers={"X-Fission-Params-UserID": "nonexistent-id"},
|
||||
json={"name": "Updated Name"},
|
||||
content_type="application/json",
|
||||
):
|
||||
import update_delete
|
||||
|
||||
result = update_delete.make_update_request()
|
||||
|
||||
assert result[1] == 404
|
||||
response_data = result[0].get_json()
|
||||
assert response_data["errorCode"] == "USER_NOT_FOUND"
|
||||
|
||||
def test_update_success(self, mocker, sample_db_row):
|
||||
"""Test successful user update returns 200.
|
||||
|
||||
Given:
|
||||
Valid X-Fission-Params-UserID header.
|
||||
User exists in database.
|
||||
Valid update data in request body.
|
||||
When:
|
||||
PUT request to update user.
|
||||
Then:
|
||||
Returns 200 OK.
|
||||
User data is updated in database.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.description = [
|
||||
MagicMock(name="id"),
|
||||
MagicMock(name="name"),
|
||||
MagicMock(name="dob"),
|
||||
MagicMock(name="email"),
|
||||
MagicMock(name="gender"),
|
||||
MagicMock(name="created"),
|
||||
MagicMock(name="modified"),
|
||||
]
|
||||
for i, col_name in enumerate(
|
||||
["id", "name", "dob", "email", "gender", "created", "modified"]
|
||||
):
|
||||
mock_cursor.description[i].name = col_name
|
||||
|
||||
# First fetchone returns existing user, second returns updated user
|
||||
mock_cursor.fetchone.side_effect = [sample_db_row, sample_db_row]
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.__enter__ = MagicMock(return_value=mock_conn)
|
||||
mock_conn.__exit__ = MagicMock(return_value=False)
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mocker.patch("update_delete.init_db_connection", return_value=mock_conn)
|
||||
|
||||
with app.test_request_context(
|
||||
method="PUT",
|
||||
headers={"X-Fission-Params-UserID": "550e8400-e29b-41d4-a716-446655440000"},
|
||||
json={"name": "Updated Name"},
|
||||
content_type="application/json",
|
||||
):
|
||||
import update_delete
|
||||
|
||||
result = update_delete.make_update_request()
|
||||
|
||||
assert result[1] == 200
|
||||
|
||||
def test_update_validation_error_invalid_email(self, mocker):
|
||||
"""Test invalid email format raises serialization error.
|
||||
|
||||
Given:
|
||||
Valid X-Fission-Params-UserID header.
|
||||
Request body with invalid email format 'invalid-email'.
|
||||
When:
|
||||
PUT request to update user.
|
||||
Then:
|
||||
Raises TypeError because ValidationError.errors() contains
|
||||
ValueError which is not JSON serializable.
|
||||
|
||||
Note:
|
||||
This test documents a bug in the source code where e.errors()
|
||||
is passed directly to jsonify without sanitization.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
with app.test_request_context(
|
||||
method="PUT",
|
||||
headers={"X-Fission-Params-UserID": "some-id"},
|
||||
json={"email": "invalid-email"},
|
||||
content_type="application/json",
|
||||
):
|
||||
import update_delete
|
||||
|
||||
with pytest.raises(TypeError, match="not JSON serializable"):
|
||||
update_delete.make_update_request()
|
||||
|
||||
def test_update_duplicate_email(self, mocker, sample_db_row):
|
||||
"""Test duplicate email on update returns 409 Conflict.
|
||||
|
||||
Given:
|
||||
Valid X-Fission-Params-UserID header.
|
||||
User exists in database.
|
||||
New email already exists for another user.
|
||||
Database raises IntegrityError on UPDATE.
|
||||
When:
|
||||
PUT request to update user email.
|
||||
Then:
|
||||
Returns 409 Conflict.
|
||||
Response contains errorCode 'DUPLICATE_USER'.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.description = [
|
||||
MagicMock(name="id"),
|
||||
MagicMock(name="name"),
|
||||
MagicMock(name="dob"),
|
||||
MagicMock(name="email"),
|
||||
MagicMock(name="gender"),
|
||||
MagicMock(name="created"),
|
||||
MagicMock(name="modified"),
|
||||
]
|
||||
for i, col_name in enumerate(
|
||||
["id", "name", "dob", "email", "gender", "created", "modified"]
|
||||
):
|
||||
mock_cursor.description[i].name = col_name
|
||||
|
||||
# First fetchone returns existing user
|
||||
mock_cursor.fetchone.return_value = sample_db_row
|
||||
# Second execute raises IntegrityError
|
||||
mock_cursor.execute.side_effect = [None, IntegrityError("duplicate key")]
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.__enter__ = MagicMock(return_value=mock_conn)
|
||||
mock_conn.__exit__ = MagicMock(return_value=False)
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mocker.patch("update_delete.init_db_connection", return_value=mock_conn)
|
||||
|
||||
with app.test_request_context(
|
||||
method="PUT",
|
||||
headers={"X-Fission-Params-UserID": "550e8400-e29b-41d4-a716-446655440000"},
|
||||
json={"email": "duplicate@example.com"},
|
||||
content_type="application/json",
|
||||
):
|
||||
import update_delete
|
||||
|
||||
result = update_delete.make_update_request()
|
||||
|
||||
assert result[1] == 409
|
||||
response_data = result[0].get_json()
|
||||
assert response_data["errorCode"] == "DUPLICATE_USER"
|
||||
|
||||
|
||||
class TestMakeDeleteRequest:
|
||||
"""Tests for make_delete_request() - user deletion."""
|
||||
|
||||
def test_delete_missing_user_id(self, mocker):
|
||||
"""Test missing X-Fission-Params-UserID header returns 400.
|
||||
|
||||
Given:
|
||||
DELETE request without X-Fission-Params-UserID header.
|
||||
When:
|
||||
make_delete_request() is called.
|
||||
Then:
|
||||
Returns 400 Bad Request.
|
||||
Response contains errorCode 'MISSING_USER_ID'.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
with app.test_request_context(method="DELETE"):
|
||||
import update_delete
|
||||
|
||||
result = update_delete.make_delete_request()
|
||||
|
||||
assert result[1] == 400
|
||||
response_data = result[0].get_json()
|
||||
assert response_data["errorCode"] == "MISSING_USER_ID"
|
||||
|
||||
def test_delete_user_not_found(self, mocker):
|
||||
"""Test delete non-existent user returns 404.
|
||||
|
||||
Given:
|
||||
Valid X-Fission-Params-UserID header.
|
||||
User does not exist in database (SELECT returns None).
|
||||
When:
|
||||
DELETE request to delete user.
|
||||
Then:
|
||||
Returns 404 Not Found.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.fetchone.return_value = None # User not found
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mocker.patch("update_delete.init_db_connection", return_value=mock_conn)
|
||||
|
||||
with app.test_request_context(
|
||||
method="DELETE",
|
||||
headers={"X-Fission-Params-UserID": "nonexistent-id"},
|
||||
):
|
||||
import update_delete
|
||||
|
||||
result = update_delete.make_delete_request()
|
||||
|
||||
assert result[1] == 404
|
||||
|
||||
def test_delete_success(self, mocker, sample_db_row):
|
||||
"""Test successful user deletion returns 200.
|
||||
|
||||
Given:
|
||||
Valid X-Fission-Params-UserID header.
|
||||
User exists in database.
|
||||
When:
|
||||
DELETE request to delete user.
|
||||
Then:
|
||||
Returns 200 OK.
|
||||
User is deleted from database.
|
||||
Response contains deleted user data.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.description = [
|
||||
MagicMock(name="id"),
|
||||
MagicMock(name="name"),
|
||||
MagicMock(name="dob"),
|
||||
MagicMock(name="email"),
|
||||
MagicMock(name="gender"),
|
||||
MagicMock(name="created"),
|
||||
MagicMock(name="modified"),
|
||||
]
|
||||
for i, col_name in enumerate(
|
||||
["id", "name", "dob", "email", "gender", "created", "modified"]
|
||||
):
|
||||
mock_cursor.description[i].name = col_name
|
||||
|
||||
# First fetchone checks existence, second returns deleted row
|
||||
mock_cursor.fetchone.side_effect = [(1,), sample_db_row]
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mocker.patch("update_delete.init_db_connection", return_value=mock_conn)
|
||||
|
||||
with app.test_request_context(
|
||||
method="DELETE",
|
||||
headers={"X-Fission-Params-UserID": "550e8400-e29b-41d4-a716-446655440000"},
|
||||
):
|
||||
import update_delete
|
||||
|
||||
result = update_delete.make_delete_request()
|
||||
|
||||
assert result[1] == 200
|
||||
|
||||
|
||||
class TestUpdatePartialFields:
|
||||
"""Tests for partial field updates."""
|
||||
|
||||
def test_update_only_name(self, mocker, sample_db_row):
|
||||
"""Test update only name field.
|
||||
|
||||
Given:
|
||||
Valid X-Fission-Params-UserID header.
|
||||
User exists in database.
|
||||
Request body contains only 'name' field.
|
||||
When:
|
||||
PUT request to update user.
|
||||
Then:
|
||||
Returns 200 OK.
|
||||
UPDATE SQL only includes name field (plus modified timestamp).
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.description = [
|
||||
MagicMock(name="id"),
|
||||
MagicMock(name="name"),
|
||||
MagicMock(name="dob"),
|
||||
MagicMock(name="email"),
|
||||
MagicMock(name="gender"),
|
||||
MagicMock(name="created"),
|
||||
MagicMock(name="modified"),
|
||||
]
|
||||
for i, col_name in enumerate(
|
||||
["id", "name", "dob", "email", "gender", "created", "modified"]
|
||||
):
|
||||
mock_cursor.description[i].name = col_name
|
||||
|
||||
mock_cursor.fetchone.side_effect = [sample_db_row, sample_db_row]
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.__enter__ = MagicMock(return_value=mock_conn)
|
||||
mock_conn.__exit__ = MagicMock(return_value=False)
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mocker.patch("update_delete.init_db_connection", return_value=mock_conn)
|
||||
|
||||
with app.test_request_context(
|
||||
method="PUT",
|
||||
headers={"X-Fission-Params-UserID": "550e8400-e29b-41d4-a716-446655440000"},
|
||||
json={"name": "New Name Only"},
|
||||
content_type="application/json",
|
||||
):
|
||||
import update_delete
|
||||
|
||||
result = update_delete.make_update_request()
|
||||
|
||||
assert result[1] == 200
|
||||
# Check that UPDATE SQL contains name field
|
||||
call_args = mock_cursor.execute.call_args_list[-1]
|
||||
sql = call_args[0][0]
|
||||
assert "name=" in sql
|
||||
|
||||
def test_update_multiple_fields(self, mocker, sample_db_row):
|
||||
"""Test update multiple fields at once.
|
||||
|
||||
Given:
|
||||
Valid X-Fission-Params-UserID header.
|
||||
User exists in database.
|
||||
Request body contains name, email, and gender fields.
|
||||
When:
|
||||
PUT request to update user.
|
||||
Then:
|
||||
Returns 200 OK.
|
||||
UPDATE SQL includes all three fields.
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.description = [
|
||||
MagicMock(name="id"),
|
||||
MagicMock(name="name"),
|
||||
MagicMock(name="dob"),
|
||||
MagicMock(name="email"),
|
||||
MagicMock(name="gender"),
|
||||
MagicMock(name="created"),
|
||||
MagicMock(name="modified"),
|
||||
]
|
||||
for i, col_name in enumerate(
|
||||
["id", "name", "dob", "email", "gender", "created", "modified"]
|
||||
):
|
||||
mock_cursor.description[i].name = col_name
|
||||
|
||||
mock_cursor.fetchone.side_effect = [sample_db_row, sample_db_row]
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.__enter__ = MagicMock(return_value=mock_conn)
|
||||
mock_conn.__exit__ = MagicMock(return_value=False)
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mocker.patch("update_delete.init_db_connection", return_value=mock_conn)
|
||||
|
||||
with app.test_request_context(
|
||||
method="PUT",
|
||||
headers={"X-Fission-Params-UserID": "550e8400-e29b-41d4-a716-446655440000"},
|
||||
json={"name": "New Name", "email": "new@example.com", "gender": "female"},
|
||||
content_type="application/json",
|
||||
):
|
||||
import update_delete
|
||||
|
||||
result = update_delete.make_update_request()
|
||||
|
||||
assert result[1] == 200
|
||||
call_args = mock_cursor.execute.call_args_list[-1]
|
||||
sql = call_args[0][0]
|
||||
assert "name=" in sql
|
||||
assert "email=" in sql
|
||||
assert "gender=" in sql
|
||||
Reference in New Issue
Block a user