410 lines
15 KiB
Python
410 lines
15 KiB
Python
import json
|
|
import tempfile
|
|
import base64
|
|
from datetime import timedelta
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.core.cache import cache
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from django.core.management import call_command
|
|
from django.core.management.base import CommandError
|
|
from django.test import TestCase
|
|
from django.utils import timezone
|
|
from rest_framework.test import APIClient, APITestCase
|
|
|
|
from adventures.models import (
|
|
Collection,
|
|
CollectionItineraryItem,
|
|
ContentImage,
|
|
Lodging,
|
|
Note,
|
|
Transportation,
|
|
)
|
|
from adventures.utils.weather import fetch_daily_temperature
|
|
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
class WeatherViewTests(APITestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(
|
|
username="weather-user",
|
|
email="weather@example.com",
|
|
password="password123",
|
|
)
|
|
self.client.force_authenticate(user=self.user)
|
|
cache.clear()
|
|
|
|
def test_daily_temperatures_rejects_too_many_days(self):
|
|
payload = {
|
|
"days": [
|
|
{"date": "2026-01-01", "latitude": 10, "longitude": 10}
|
|
for _ in range(61)
|
|
]
|
|
}
|
|
|
|
response = self.client.post(
|
|
"/api/weather/daily-temperatures/", payload, format="json"
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn("maximum", response.json().get("error", "").lower())
|
|
|
|
@patch("adventures.views.weather_view.WeatherViewSet._fetch_daily_temperature")
|
|
def test_daily_temperatures_future_date_reaches_fetch_path(
|
|
self, mock_fetch_temperature
|
|
):
|
|
future_date = (timezone.now().date() + timedelta(days=10)).isoformat()
|
|
mock_fetch_temperature.return_value = {
|
|
"date": future_date,
|
|
"available": True,
|
|
"temperature_low_c": 19.0,
|
|
"temperature_high_c": 26.0,
|
|
"temperature_c": 22.5,
|
|
"is_estimate": False,
|
|
"source": "forecast",
|
|
}
|
|
|
|
response = self.client.post(
|
|
"/api/weather/daily-temperatures/",
|
|
{"days": [{"date": future_date, "latitude": 12.34, "longitude": 56.78}]},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()["results"][0]["date"], future_date)
|
|
self.assertTrue(response.json()["results"][0]["available"])
|
|
self.assertEqual(response.json()["results"][0]["temperature_low_c"], 19.0)
|
|
self.assertEqual(response.json()["results"][0]["temperature_high_c"], 26.0)
|
|
self.assertFalse(response.json()["results"][0]["is_estimate"])
|
|
self.assertEqual(response.json()["results"][0]["source"], "forecast")
|
|
self.assertEqual(response.json()["results"][0]["temperature_c"], 22.5)
|
|
mock_fetch_temperature.assert_called_once_with(future_date, 12.34, 56.78)
|
|
|
|
@patch("adventures.utils.weather.requests.get")
|
|
def test_daily_temperatures_far_future_uses_historical_estimate(
|
|
self, mock_requests_get
|
|
):
|
|
future_date = (timezone.now().date() + timedelta(days=3650)).isoformat()
|
|
|
|
archive_no_data = Mock()
|
|
archive_no_data.raise_for_status.return_value = None
|
|
archive_no_data.json.return_value = {"daily": {}}
|
|
|
|
forecast_no_data = Mock()
|
|
forecast_no_data.raise_for_status.return_value = None
|
|
forecast_no_data.json.return_value = {"daily": {}}
|
|
|
|
historical_data = Mock()
|
|
historical_data.raise_for_status.return_value = None
|
|
historical_data.json.return_value = {
|
|
"daily": {
|
|
"temperature_2m_max": [15.0, 18.0, 20.0],
|
|
"temperature_2m_min": [7.0, 9.0, 11.0],
|
|
}
|
|
}
|
|
|
|
call_sequence = [archive_no_data, forecast_no_data, historical_data]
|
|
|
|
def mock_get(*args, **kwargs):
|
|
if call_sequence:
|
|
return call_sequence.pop(0)
|
|
return historical_data
|
|
|
|
mock_requests_get.side_effect = mock_get
|
|
|
|
response = self.client.post(
|
|
"/api/weather/daily-temperatures/",
|
|
{"days": [{"date": future_date, "latitude": 12.34, "longitude": 56.78}]},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
result = response.json()["results"][0]
|
|
self.assertTrue(result["available"])
|
|
self.assertEqual(result["date"], future_date)
|
|
self.assertEqual(result["temperature_low_c"], 9.0)
|
|
self.assertEqual(result["temperature_high_c"], 17.7)
|
|
self.assertEqual(result["temperature_c"], 13.3)
|
|
self.assertTrue(result["is_estimate"])
|
|
self.assertEqual(result["source"], "historical_estimate")
|
|
self.assertGreaterEqual(mock_requests_get.call_count, 3)
|
|
|
|
@patch("adventures.utils.weather.requests.get")
|
|
def test_daily_temperatures_accepts_zero_lat_lon(self, mock_requests_get):
|
|
today = timezone.now().date().isoformat()
|
|
mocked_response = Mock()
|
|
mocked_response.raise_for_status.return_value = None
|
|
mocked_response.json.return_value = {
|
|
"daily": {
|
|
"temperature_2m_max": [20.0],
|
|
"temperature_2m_min": [10.0],
|
|
}
|
|
}
|
|
mock_requests_get.return_value = mocked_response
|
|
|
|
response = self.client.post(
|
|
"/api/weather/daily-temperatures/",
|
|
{"days": [{"date": today, "latitude": 0, "longitude": 0}]},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()["results"][0]["date"], today)
|
|
self.assertTrue(response.json()["results"][0]["available"])
|
|
self.assertEqual(response.json()["results"][0]["temperature_low_c"], 10.0)
|
|
self.assertEqual(response.json()["results"][0]["temperature_high_c"], 20.0)
|
|
self.assertFalse(response.json()["results"][0]["is_estimate"])
|
|
self.assertEqual(response.json()["results"][0]["source"], "archive")
|
|
self.assertEqual(response.json()["results"][0]["temperature_c"], 15.0)
|
|
|
|
|
|
class WeatherHelperTests(TestCase):
|
|
@patch("adventures.utils.weather.requests.get")
|
|
def test_fetch_daily_temperature_returns_unavailable_when_all_sources_fail(
|
|
self, mock_requests_get
|
|
):
|
|
mocked_response = Mock()
|
|
mocked_response.raise_for_status.return_value = None
|
|
mocked_response.json.return_value = {"daily": {}}
|
|
mock_requests_get.return_value = mocked_response
|
|
|
|
result = fetch_daily_temperature(
|
|
date=(timezone.now().date() + timedelta(days=6000)).isoformat(),
|
|
latitude=40.7128,
|
|
longitude=-74.0060,
|
|
)
|
|
|
|
self.assertEqual(
|
|
result,
|
|
{
|
|
"date": result["date"],
|
|
"available": False,
|
|
"temperature_low_c": None,
|
|
"temperature_high_c": None,
|
|
"temperature_c": None,
|
|
"is_estimate": False,
|
|
"source": None,
|
|
},
|
|
)
|
|
|
|
|
|
class MCPAuthTests(APITestCase):
|
|
def test_mcp_unauthenticated_access_is_rejected(self):
|
|
unauthenticated_client = APIClient()
|
|
response = unauthenticated_client.post("/api/mcp", {}, format="json")
|
|
self.assertIn(response.status_code, [401, 403])
|
|
|
|
|
|
class LocationPayloadHardeningTests(APITestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(
|
|
username="location-hardening-user",
|
|
email="location-hardening@example.com",
|
|
password="password123",
|
|
)
|
|
self.client.force_authenticate(user=self.user)
|
|
|
|
def test_create_location_truncates_overlong_name_and_location(self):
|
|
overlong_name = "N" * 250
|
|
overlong_location = "L" * 250
|
|
|
|
response = self.client.post(
|
|
"/api/locations/",
|
|
{
|
|
"name": overlong_name,
|
|
"location": overlong_location,
|
|
"is_public": False,
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 201)
|
|
self.assertEqual(len(response.data["name"]), 200)
|
|
self.assertEqual(len(response.data["location"]), 200)
|
|
self.assertEqual(response.data["name"], overlong_name[:200])
|
|
self.assertEqual(response.data["location"], overlong_location[:200])
|
|
|
|
def test_create_location_accepts_high_precision_coordinates(self):
|
|
response = self.client.post(
|
|
"/api/locations/",
|
|
{
|
|
"name": "Precision test",
|
|
"is_public": False,
|
|
"latitude": 51.5007292,
|
|
"longitude": -0.1246254,
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 201)
|
|
self.assertEqual(response.data["latitude"], "51.500729")
|
|
self.assertEqual(response.data["longitude"], "-0.124625")
|
|
|
|
|
|
class CollectionViewSetTests(APITestCase):
|
|
def setUp(self):
|
|
self.owner = User.objects.create_user(
|
|
username="collection-owner",
|
|
email="owner@example.com",
|
|
password="password123",
|
|
)
|
|
self.shared_user = User.objects.create_user(
|
|
username="collection-shared",
|
|
email="shared@example.com",
|
|
password="password123",
|
|
)
|
|
|
|
def _create_test_image_file(self, name="test.png"):
|
|
# 1x1 PNG
|
|
png_bytes = base64.b64decode(
|
|
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO7Y9x8AAAAASUVORK5CYII="
|
|
)
|
|
return SimpleUploadedFile(name, png_bytes, content_type="image/png")
|
|
|
|
def _create_collection_with_non_location_images(self):
|
|
collection = Collection.objects.create(
|
|
user=self.owner,
|
|
name="Image fallback collection",
|
|
)
|
|
|
|
lodging = Lodging.objects.create(
|
|
user=self.owner,
|
|
collection=collection,
|
|
name="Fallback lodge",
|
|
)
|
|
transportation = Transportation.objects.create(
|
|
user=self.owner,
|
|
collection=collection,
|
|
type="car",
|
|
name="Fallback ride",
|
|
)
|
|
|
|
lodging_content_type = ContentType.objects.get_for_model(Lodging)
|
|
transportation_content_type = ContentType.objects.get_for_model(Transportation)
|
|
|
|
ContentImage.objects.create(
|
|
user=self.owner,
|
|
content_type=lodging_content_type,
|
|
object_id=lodging.id,
|
|
image=self._create_test_image_file("lodging.png"),
|
|
is_primary=True,
|
|
)
|
|
ContentImage.objects.create(
|
|
user=self.owner,
|
|
content_type=transportation_content_type,
|
|
object_id=transportation.id,
|
|
image=self._create_test_image_file("transport.png"),
|
|
is_primary=True,
|
|
)
|
|
|
|
return collection
|
|
|
|
def test_list_includes_lodging_transportation_images_when_no_location_images(self):
|
|
collection = self._create_collection_with_non_location_images()
|
|
|
|
self.client.force_authenticate(user=self.owner)
|
|
response = self.client.get("/api/collections/")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertGreater(len(response.data.get("results", [])), 0)
|
|
|
|
collection_payload = next(
|
|
item
|
|
for item in response.data["results"]
|
|
if item["id"] == str(collection.id)
|
|
)
|
|
self.assertIn("location_images", collection_payload)
|
|
self.assertGreater(len(collection_payload["location_images"]), 0)
|
|
self.assertTrue(
|
|
any(
|
|
image.get("is_primary")
|
|
for image in collection_payload["location_images"]
|
|
)
|
|
)
|
|
|
|
def test_shared_endpoint_includes_non_location_primary_images(self):
|
|
collection = self._create_collection_with_non_location_images()
|
|
collection.shared_with.add(self.shared_user)
|
|
|
|
self.client.force_authenticate(user=self.shared_user)
|
|
response = self.client.get("/api/collections/shared/")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertGreater(len(response.data), 0)
|
|
|
|
collection_payload = next(
|
|
item for item in response.data if item["id"] == str(collection.id)
|
|
)
|
|
self.assertEqual(str(collection.id), collection_payload["id"])
|
|
self.assertIn("location_images", collection_payload)
|
|
self.assertGreater(len(collection_payload["location_images"]), 0)
|
|
first_image = collection_payload["location_images"][0]
|
|
self.assertSetEqual(
|
|
set(first_image.keys()),
|
|
{"id", "image", "is_primary", "user", "immich_id"},
|
|
)
|
|
|
|
|
|
class ExportCollectionsBackupCommandTests(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(
|
|
username="backup-user",
|
|
email="backup@example.com",
|
|
password="password123",
|
|
)
|
|
self.collaborator = User.objects.create_user(
|
|
username="collab-user",
|
|
email="collab@example.com",
|
|
password="password123",
|
|
)
|
|
self.collection = Collection.objects.create(
|
|
user=self.user,
|
|
name="My Trip",
|
|
description="Backup test collection",
|
|
)
|
|
self.collection.shared_with.add(self.collaborator)
|
|
|
|
note = Note.objects.create(user=self.user, name="Test item")
|
|
note_content_type = ContentType.objects.get_for_model(Note)
|
|
CollectionItineraryItem.objects.create(
|
|
collection=self.collection,
|
|
content_type=note_content_type,
|
|
object_id=note.id,
|
|
date=timezone.now().date(),
|
|
is_global=False,
|
|
order=1,
|
|
)
|
|
|
|
def test_export_collections_backup_writes_expected_payload(self):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
output_file = Path(temp_dir) / "collections_snapshot.json"
|
|
|
|
call_command("export_collections_backup", output=str(output_file))
|
|
|
|
self.assertTrue(output_file.exists())
|
|
payload = json.loads(output_file.read_text(encoding="utf-8"))
|
|
|
|
self.assertEqual(payload["backup_type"], "collections_snapshot")
|
|
self.assertIn("timestamp", payload)
|
|
self.assertEqual(payload["counts"]["collections"], 1)
|
|
self.assertEqual(payload["counts"]["collection_itinerary_items"], 1)
|
|
self.assertEqual(len(payload["collections"]), 1)
|
|
self.assertEqual(len(payload["collection_itinerary_items"]), 1)
|
|
self.assertEqual(
|
|
payload["collections"][0]["shared_with_ids"],
|
|
[self.collaborator.id],
|
|
)
|
|
|
|
def test_export_collections_backup_raises_for_missing_output_directory(self):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
missing_directory = Path(temp_dir) / "missing"
|
|
output_file = missing_directory / "collections_snapshot.json"
|
|
|
|
with self.assertRaises(CommandError):
|
|
call_command("export_collections_backup", output=str(output_file))
|