forked from enviPath/enviPy
Tobias O
5789f20e7f
Add API key authentication to v1 API Also includes: - management command to create keys for users - Improvements to API tests Minor: - more robust way to start docker dev container. Reviewed-on: enviPath/enviPy#327 Co-authored-by: Tobias O <tobias.olenyi@envipath.com> Co-committed-by: Tobias O <tobias.olenyi@envipath.com>
449 lines
18 KiB
Python
449 lines
18 KiB
Python
"""
|
|
Tests for Additional Information API endpoints.
|
|
|
|
Tests CRUD operations on scenario additional information including the new PATCH endpoint.
|
|
"""
|
|
|
|
from django.test import TestCase, tag
|
|
import json
|
|
from uuid import uuid4
|
|
|
|
from epdb.logic import PackageManager, UserManager
|
|
from epdb.models import Scenario
|
|
|
|
|
|
@tag("api", "additional_information")
|
|
class AdditionalInformationAPITests(TestCase):
|
|
"""Test additional information API endpoints."""
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
"""Set up test data: user, package, and scenario."""
|
|
cls.user = UserManager.create_user(
|
|
"ai-test-user",
|
|
"ai-test@envipath.com",
|
|
"SuperSafe",
|
|
set_setting=False,
|
|
add_to_group=False,
|
|
is_active=True,
|
|
)
|
|
cls.other_user = UserManager.create_user(
|
|
"ai-other-user",
|
|
"ai-other@envipath.com",
|
|
"SuperSafe",
|
|
set_setting=False,
|
|
add_to_group=False,
|
|
is_active=True,
|
|
)
|
|
cls.package = PackageManager.create_package(
|
|
cls.user, "AI Test Package", "Test package for additional information"
|
|
)
|
|
# Package owned by other_user (no access for cls.user)
|
|
cls.other_package = PackageManager.create_package(
|
|
cls.other_user, "Other Package", "Package without access"
|
|
)
|
|
# Create a scenario for testing
|
|
cls.scenario = Scenario.objects.create(
|
|
package=cls.package,
|
|
name="Test Scenario",
|
|
description="Test scenario for additional information tests",
|
|
scenario_type="biodegradation",
|
|
scenario_date="2024-01-01",
|
|
additional_information={}, # Initialize with empty dict
|
|
)
|
|
cls.other_scenario = Scenario.objects.create(
|
|
package=cls.other_package,
|
|
name="Other Scenario",
|
|
description="Scenario in package without access",
|
|
scenario_type="biodegradation",
|
|
scenario_date="2024-01-01",
|
|
additional_information={},
|
|
)
|
|
|
|
def test_list_all_schemas(self):
|
|
"""Test GET /api/v1/information/schema/ returns all schemas."""
|
|
self.client.force_login(self.user)
|
|
|
|
response = self.client.get("/api/v1/information/schema/")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
self.assertIsInstance(data, dict)
|
|
# Should have multiple schemas
|
|
self.assertGreater(len(data), 0)
|
|
# Each schema should have RJSF format
|
|
for name, schema in data.items():
|
|
self.assertIn("schema", schema)
|
|
self.assertIn("uiSchema", schema)
|
|
self.assertIn("formData", schema)
|
|
self.assertIn("groups", schema)
|
|
|
|
def test_get_specific_schema(self):
|
|
"""Test GET /api/v1/information/schema/{model_name}/ returns specific schema."""
|
|
self.client.force_login(self.user)
|
|
|
|
# Assuming 'temperature' is a valid model
|
|
response = self.client.get("/api/v1/information/schema/temperature/")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
self.assertIn("schema", data)
|
|
self.assertIn("uiSchema", data)
|
|
|
|
def test_get_nonexistent_schema_returns_404(self):
|
|
"""Test GET for non-existent schema returns 404."""
|
|
self.client.force_login(self.user)
|
|
|
|
response = self.client.get("/api/v1/information/schema/nonexistent/")
|
|
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_list_scenario_information_empty(self):
|
|
"""Test GET /api/v1/scenario/{uuid}/information/ returns empty list initially."""
|
|
self.client.force_login(self.user)
|
|
|
|
response = self.client.get(f"/api/v1/scenario/{self.scenario.uuid}/information/")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
self.assertIsInstance(data, list)
|
|
self.assertEqual(len(data), 0)
|
|
|
|
def test_create_additional_information(self):
|
|
"""Test POST creates additional information."""
|
|
self.client.force_login(self.user)
|
|
|
|
# Create temperature information (assuming temperature model exists)
|
|
payload = {"interval": {"start": 20, "end": 25}}
|
|
response = self.client.post(
|
|
f"/api/v1/scenario/{self.scenario.uuid}/information/temperature/",
|
|
data=json.dumps(payload),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
self.assertEqual(data["status"], "created")
|
|
self.assertIn("uuid", data)
|
|
self.assertIsNotNone(data["uuid"])
|
|
|
|
def test_create_with_invalid_data_returns_400(self):
|
|
"""Test POST with invalid data returns 400 with validation errors."""
|
|
self.client.force_login(self.user)
|
|
|
|
# Invalid data (missing required fields or wrong types)
|
|
payload = {"invalid_field": "value"}
|
|
response = self.client.post(
|
|
f"/api/v1/scenario/{self.scenario.uuid}/information/temperature/",
|
|
data=json.dumps(payload),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 400)
|
|
data = response.json()
|
|
# Should have validation error details in 'detail' field
|
|
self.assertIn("detail", data)
|
|
|
|
def test_validation_errors_are_user_friendly(self):
|
|
"""Test that validation errors are user-friendly and field-specific."""
|
|
self.client.force_login(self.user)
|
|
|
|
# Invalid data - wrong type (string instead of number in interval)
|
|
payload = {"interval": {"start": "not_a_number", "end": 25}}
|
|
response = self.client.post(
|
|
f"/api/v1/scenario/{self.scenario.uuid}/information/temperature/",
|
|
data=json.dumps(payload),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 400)
|
|
data = response.json()
|
|
|
|
# Parse the error response - Django Ninja wraps errors in 'detail'
|
|
error_str = data.get("detail") or data.get("error")
|
|
self.assertIsNotNone(error_str, "Response should contain error details")
|
|
|
|
# Parse the JSON error string
|
|
error_data = json.loads(error_str)
|
|
|
|
# Check structure
|
|
self.assertEqual(error_data.get("type"), "validation_error")
|
|
self.assertIn("field_errors", error_data)
|
|
self.assertIn("message", error_data)
|
|
|
|
# Ensure error messages are user-friendly (no Pydantic URLs or technical jargon)
|
|
error_str = json.dumps(error_data)
|
|
self.assertNotIn("pydantic", error_str.lower())
|
|
self.assertNotIn("https://errors.pydantic.dev", error_str)
|
|
self.assertNotIn("loc", error_str) # No technical field like 'loc'
|
|
|
|
# Check that error message is helpful
|
|
self.assertIn("Please", error_data["message"]) # User-friendly language
|
|
|
|
def test_patch_additional_information(self):
|
|
"""Test PATCH updates existing additional information."""
|
|
self.client.force_login(self.user)
|
|
|
|
# First create an item
|
|
create_payload = {"interval": {"start": 20, "end": 25}}
|
|
create_response = self.client.post(
|
|
f"/api/v1/scenario/{self.scenario.uuid}/information/temperature/",
|
|
data=json.dumps(create_payload),
|
|
content_type="application/json",
|
|
)
|
|
item_uuid = create_response.json()["uuid"]
|
|
|
|
# Then update it with PATCH
|
|
update_payload = {"interval": {"start": 30, "end": 35}}
|
|
patch_response = self.client.patch(
|
|
f"/api/v1/scenario/{self.scenario.uuid}/information/item/{item_uuid}/",
|
|
data=json.dumps(update_payload),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(patch_response.status_code, 200)
|
|
data = patch_response.json()
|
|
self.assertEqual(data["status"], "updated")
|
|
self.assertEqual(data["uuid"], item_uuid) # UUID preserved
|
|
|
|
# Verify the data was updated
|
|
list_response = self.client.get(f"/api/v1/scenario/{self.scenario.uuid}/information/")
|
|
items = list_response.json()
|
|
self.assertEqual(len(items), 1)
|
|
updated_item = items[0]
|
|
self.assertEqual(updated_item["uuid"], item_uuid)
|
|
self.assertEqual(updated_item["data"]["interval"]["start"], 30)
|
|
self.assertEqual(updated_item["data"]["interval"]["end"], 35)
|
|
|
|
def test_patch_nonexistent_item_returns_404(self):
|
|
"""Test PATCH on non-existent item returns 404."""
|
|
self.client.force_login(self.user)
|
|
|
|
fake_uuid = str(uuid4())
|
|
payload = {"interval": {"start": 30, "end": 35}}
|
|
response = self.client.patch(
|
|
f"/api/v1/scenario/{self.scenario.uuid}/information/item/{fake_uuid}/",
|
|
data=json.dumps(payload),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_patch_with_invalid_data_returns_400(self):
|
|
"""Test PATCH with invalid data returns 400."""
|
|
self.client.force_login(self.user)
|
|
|
|
# First create an item
|
|
create_payload = {"interval": {"start": 20, "end": 25}}
|
|
create_response = self.client.post(
|
|
f"/api/v1/scenario/{self.scenario.uuid}/information/temperature/",
|
|
data=json.dumps(create_payload),
|
|
content_type="application/json",
|
|
)
|
|
item_uuid = create_response.json()["uuid"]
|
|
|
|
# Try to update with invalid data
|
|
invalid_payload = {"invalid_field": "value"}
|
|
patch_response = self.client.patch(
|
|
f"/api/v1/scenario/{self.scenario.uuid}/information/item/{item_uuid}/",
|
|
data=json.dumps(invalid_payload),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(patch_response.status_code, 400)
|
|
|
|
def test_patch_validation_errors_are_user_friendly(self):
|
|
"""Test that PATCH validation errors are user-friendly and field-specific."""
|
|
self.client.force_login(self.user)
|
|
|
|
# First create an item
|
|
create_payload = {"interval": {"start": 20, "end": 25}}
|
|
create_response = self.client.post(
|
|
f"/api/v1/scenario/{self.scenario.uuid}/information/temperature/",
|
|
data=json.dumps(create_payload),
|
|
content_type="application/json",
|
|
)
|
|
item_uuid = create_response.json()["uuid"]
|
|
|
|
# Update with invalid data - wrong type (string instead of number in interval)
|
|
invalid_payload = {"interval": {"start": "not_a_number", "end": 25}}
|
|
patch_response = self.client.patch(
|
|
f"/api/v1/scenario/{self.scenario.uuid}/information/item/{item_uuid}/",
|
|
data=json.dumps(invalid_payload),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(patch_response.status_code, 400)
|
|
data = patch_response.json()
|
|
|
|
# Parse the error response - Django Ninja wraps errors in 'detail'
|
|
error_str = data.get("detail") or data.get("error")
|
|
self.assertIsNotNone(error_str, "Response should contain error details")
|
|
|
|
# Parse the JSON error string
|
|
error_data = json.loads(error_str)
|
|
|
|
# Check structure
|
|
self.assertEqual(error_data.get("type"), "validation_error")
|
|
self.assertIn("field_errors", error_data)
|
|
self.assertIn("message", error_data)
|
|
|
|
# Ensure error messages are user-friendly (no Pydantic URLs or technical jargon)
|
|
error_str = json.dumps(error_data)
|
|
self.assertNotIn("pydantic", error_str.lower())
|
|
self.assertNotIn("https://errors.pydantic.dev", error_str)
|
|
self.assertNotIn("loc", error_str) # No technical field like 'loc'
|
|
|
|
# Check that error message is helpful
|
|
self.assertIn("Please", error_data["message"]) # User-friendly language
|
|
|
|
def test_delete_additional_information(self):
|
|
"""Test DELETE removes additional information."""
|
|
self.client.force_login(self.user)
|
|
|
|
# Create an item
|
|
create_payload = {"interval": {"start": 20, "end": 25}}
|
|
create_response = self.client.post(
|
|
f"/api/v1/scenario/{self.scenario.uuid}/information/temperature/",
|
|
data=json.dumps(create_payload),
|
|
content_type="application/json",
|
|
)
|
|
item_uuid = create_response.json()["uuid"]
|
|
|
|
# Delete it
|
|
delete_response = self.client.delete(
|
|
f"/api/v1/scenario/{self.scenario.uuid}/information/item/{item_uuid}/"
|
|
)
|
|
|
|
self.assertEqual(delete_response.status_code, 200)
|
|
data = delete_response.json()
|
|
self.assertEqual(data["status"], "deleted")
|
|
|
|
# Verify deletion
|
|
list_response = self.client.get(f"/api/v1/scenario/{self.scenario.uuid}/information/")
|
|
items = list_response.json()
|
|
self.assertEqual(len(items), 0)
|
|
|
|
def test_delete_nonexistent_item_returns_404(self):
|
|
"""Test DELETE on non-existent item returns 404."""
|
|
self.client.force_login(self.user)
|
|
|
|
fake_uuid = str(uuid4())
|
|
response = self.client.delete(
|
|
f"/api/v1/scenario/{self.scenario.uuid}/information/item/{fake_uuid}/"
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_multiple_items_crud(self):
|
|
"""Test creating, updating, and deleting multiple items."""
|
|
self.client.force_login(self.user)
|
|
|
|
# Create first item
|
|
item1_payload = {"interval": {"start": 20, "end": 25}}
|
|
response1 = self.client.post(
|
|
f"/api/v1/scenario/{self.scenario.uuid}/information/temperature/",
|
|
data=json.dumps(item1_payload),
|
|
content_type="application/json",
|
|
)
|
|
item1_uuid = response1.json()["uuid"]
|
|
|
|
# Create second item (different type if available, or same type)
|
|
item2_payload = {"interval": {"start": 30, "end": 35}}
|
|
response2 = self.client.post(
|
|
f"/api/v1/scenario/{self.scenario.uuid}/information/temperature/",
|
|
data=json.dumps(item2_payload),
|
|
content_type="application/json",
|
|
)
|
|
item2_uuid = response2.json()["uuid"]
|
|
|
|
# Verify both exist
|
|
list_response = self.client.get(f"/api/v1/scenario/{self.scenario.uuid}/information/")
|
|
items = list_response.json()
|
|
self.assertEqual(len(items), 2)
|
|
|
|
# Update first item
|
|
update_payload = {"interval": {"start": 15, "end": 20}}
|
|
self.client.patch(
|
|
f"/api/v1/scenario/{self.scenario.uuid}/information/item/{item1_uuid}/",
|
|
data=json.dumps(update_payload),
|
|
content_type="application/json",
|
|
)
|
|
|
|
# Delete second item
|
|
self.client.delete(f"/api/v1/scenario/{self.scenario.uuid}/information/item/{item2_uuid}/")
|
|
|
|
# Verify final state: one item with updated data
|
|
list_response = self.client.get(f"/api/v1/scenario/{self.scenario.uuid}/information/")
|
|
items = list_response.json()
|
|
self.assertEqual(len(items), 1)
|
|
self.assertEqual(items[0]["uuid"], item1_uuid)
|
|
self.assertEqual(items[0]["data"]["interval"]["start"], 15)
|
|
|
|
def test_list_info_denied_without_permission(self):
|
|
"""User cannot list info for scenario in package they don't have access to"""
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(f"/api/v1/scenario/{self.other_scenario.uuid}/information/")
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
def test_add_info_denied_without_permission(self):
|
|
"""User cannot add info to scenario in package they don't have access to"""
|
|
self.client.force_login(self.user)
|
|
payload = {"interval": {"start": 25, "end": 30}}
|
|
response = self.client.post(
|
|
f"/api/v1/scenario/{self.other_scenario.uuid}/information/temperature/",
|
|
json.dumps(payload),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
def test_update_info_denied_without_permission(self):
|
|
"""User cannot update info in scenario they don't have access to"""
|
|
self.client.force_login(self.other_user)
|
|
# First create an item as other_user
|
|
create_payload = {"interval": {"start": 20, "end": 25}}
|
|
create_response = self.client.post(
|
|
f"/api/v1/scenario/{self.other_scenario.uuid}/information/temperature/",
|
|
data=json.dumps(create_payload),
|
|
content_type="application/json",
|
|
)
|
|
item_uuid = create_response.json()["uuid"]
|
|
|
|
# Try to update as user (who doesn't have access)
|
|
self.client.force_login(self.user)
|
|
update_payload = {"interval": {"start": 30, "end": 35}}
|
|
response = self.client.patch(
|
|
f"/api/v1/scenario/{self.other_scenario.uuid}/information/item/{item_uuid}/",
|
|
data=json.dumps(update_payload),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
def test_delete_info_denied_without_permission(self):
|
|
"""User cannot delete info from scenario they don't have access to"""
|
|
self.client.force_login(self.other_user)
|
|
# First create an item as other_user
|
|
create_payload = {"interval": {"start": 20, "end": 25}}
|
|
create_response = self.client.post(
|
|
f"/api/v1/scenario/{self.other_scenario.uuid}/information/temperature/",
|
|
data=json.dumps(create_payload),
|
|
content_type="application/json",
|
|
)
|
|
item_uuid = create_response.json()["uuid"]
|
|
|
|
# Try to delete as user (who doesn't have access)
|
|
self.client.force_login(self.user)
|
|
response = self.client.delete(
|
|
f"/api/v1/scenario/{self.other_scenario.uuid}/information/item/{item_uuid}/"
|
|
)
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
def test_nonexistent_scenario_returns_404(self):
|
|
"""Test operations on non-existent scenario return 404."""
|
|
self.client.force_login(self.user)
|
|
|
|
fake_uuid = uuid4()
|
|
response = self.client.get(f"/api/v1/scenario/{fake_uuid}/information/")
|
|
|
|
self.assertEqual(response.status_code, 404)
|