From b0dfb35cf2c8a0911c2562039471398225675ce5 Mon Sep 17 00:00:00 2001 From: Jan Date: Sat, 5 Sep 2026 19:10:44 +0200 Subject: [PATCH 1/2] feat(tests): increase coverage --- backend/tests/analytics/api/conftest.py | 27 ++ .../test_analytics_market_insight_viewset.py | 116 ++++++ .../test_analytics_plan_aggregate_viewset.py | 120 +++++++ backend/tests/gamedata/api/conftest.py | 16 + .../gamedata/api/test_gamedata_viewsets.py | 162 +++++++++ .../planning/api/viewsets/test_cx_viewset.py | 141 ++++++++ .../api/viewsets/test_empire_viewset.py | 216 ++++++++++++ .../api/viewsets/test_plan_viewset.py | 128 +++++++ backend/tests/planning/conftest.py | 10 + backend/tests/user/api/test_user_viewsets.py | 331 ++++++++++++++++++ 10 files changed, 1267 insertions(+) create mode 100644 backend/tests/analytics/api/conftest.py create mode 100644 backend/tests/analytics/api/test_analytics_market_insight_viewset.py create mode 100644 backend/tests/analytics/api/test_analytics_plan_aggregate_viewset.py create mode 100644 backend/tests/gamedata/api/conftest.py create mode 100644 backend/tests/gamedata/api/test_gamedata_viewsets.py create mode 100644 backend/tests/planning/api/viewsets/test_cx_viewset.py create mode 100644 backend/tests/planning/api/viewsets/test_empire_viewset.py create mode 100644 backend/tests/planning/api/viewsets/test_plan_viewset.py create mode 100644 backend/tests/user/api/test_user_viewsets.py diff --git a/backend/tests/analytics/api/conftest.py b/backend/tests/analytics/api/conftest.py new file mode 100644 index 0000000..6c47a58 --- /dev/null +++ b/backend/tests/analytics/api/conftest.py @@ -0,0 +1,27 @@ +from collections.abc import Callable + +import pytest +from analytics.models import AnalyticsEmpireMaterialSnapshot, AnalyticsPlanAggregate +from gamedata.models.game_planet import GamePlanet +from model_bakery import baker +from planning.models import PlanningEmpire + + +@pytest.fixture() +def planet_factory() -> Callable[..., GamePlanet]: + return lambda **kwargs: baker.make('gamedata.GamePlanet', make_m2m=True, **kwargs) + + +@pytest.fixture() +def plan_aggregate_factory() -> Callable[..., AnalyticsPlanAggregate]: + return lambda **kwargs: baker.make('analytics.AnalyticsPlanAggregate', **kwargs) + + +@pytest.fixture() +def empire_factory() -> Callable[..., PlanningEmpire]: + return lambda **kwargs: baker.make('planning.PlanningEmpire', **kwargs) + + +@pytest.fixture() +def material_snapshot_factory() -> Callable[..., AnalyticsEmpireMaterialSnapshot]: + return lambda **kwargs: baker.make('analytics.AnalyticsEmpireMaterialSnapshot', **kwargs) diff --git a/backend/tests/analytics/api/test_analytics_market_insight_viewset.py b/backend/tests/analytics/api/test_analytics_market_insight_viewset.py new file mode 100644 index 0000000..e08464f --- /dev/null +++ b/backend/tests/analytics/api/test_analytics_market_insight_viewset.py @@ -0,0 +1,116 @@ +from collections.abc import Callable +from datetime import timedelta + +import pytest +from analytics.models import AnalyticsEmpireMaterialSnapshot +from django.urls import reverse +from django.utils import timezone +from planning.models import PlanningEmpire +from rest_framework.test import APIClient +from user.models import User + +pytestmark = pytest.mark.django_db + + +def _url() -> str: + return reverse('analytics:planning-insight-materials') + + +def _rows_by_ticker(data: list[list[object]]) -> dict[object, list[object]]: + return {row[0]: row for row in data} + + +class TestAnalyticsMarketInsightViewSetAccess: + def test_unauthenticated_access_is_allowed(self, api_client: APIClient) -> None: + response = api_client.get(_url()) + + assert response.status_code == 200 + + def test_authenticated_access_also_succeeds( + self, api_client: APIClient, user_factory: Callable[..., User] + ) -> None: + user = user_factory() + + response = api_client.as_user(user).get(_url()) # ty:ignore[unresolved-attribute] + + assert response.status_code == 200 + + def test_disallowed_http_method_returns_405(self, api_client: APIClient) -> None: + response = api_client.post(_url(), data={}) + + assert response.status_code == 405 + + +class TestAnalyticsMarketInsightViewSetGlobalMaterials: + def test_empty_database_returns_empty_list(self, api_client: APIClient) -> None: + assert AnalyticsEmpireMaterialSnapshot.objects.count() == 0 + + response = api_client.get(_url()) + + assert response.status_code == 200 + assert response.data == [] + + def test_aggregates_production_consumption_and_delta_per_material( + self, + api_client: APIClient, + empire_factory: Callable[..., PlanningEmpire], + material_snapshot_factory: Callable[..., AnalyticsEmpireMaterialSnapshot], + ) -> None: + empire_a = empire_factory() + empire_b = empire_factory() + + material_snapshot_factory( + empire=empire_a, material_ticker='H2O', production=100, consumption=40, delta=60 + ) + material_snapshot_factory( + empire=empire_b, material_ticker='H2O', production=50, consumption=10, delta=40 + ) + material_snapshot_factory( + empire=empire_a, material_ticker='DW', production=5, consumption=5, delta=0 + ) + + response = api_client.get(_url()) + + assert response.status_code == 200 + rows = _rows_by_ticker(response.data) + + assert rows['H2O'] == ['H2O', 150.0, 50.0, 100.0] + assert rows['DW'] == ['DW', 5.0, 5.0, 0.0] + + def test_results_are_ordered_by_material_ticker( + self, + api_client: APIClient, + empire_factory: Callable[..., PlanningEmpire], + material_snapshot_factory: Callable[..., AnalyticsEmpireMaterialSnapshot], + ) -> None: + empire = empire_factory() + + material_snapshot_factory(empire=empire, material_ticker='RAT') + material_snapshot_factory(empire=empire, material_ticker='DW') + material_snapshot_factory(empire=empire, material_ticker='H2O') + + response = api_client.get(_url()) + + tickers = [row[0] for row in response.data] + assert tickers == ['DW', 'H2O', 'RAT'] + + def test_excludes_snapshots_from_empires_inactive_for_over_30_days( + self, + api_client: APIClient, + empire_factory: Callable[..., PlanningEmpire], + material_snapshot_factory: Callable[..., AnalyticsEmpireMaterialSnapshot], + ) -> None: + active_empire = empire_factory() + stale_empire = empire_factory() + + material_snapshot_factory(empire=active_empire, material_ticker='H2O', production=10, consumption=0, delta=10) + material_snapshot_factory(empire=stale_empire, material_ticker='H2O', production=999, consumption=0, delta=999) + + stale_cutoff = timezone.now() - timedelta(days=40) + PlanningEmpire.objects.filter(pk=stale_empire.pk).update(modified_at=stale_cutoff) + + response = api_client.get(_url()) + + assert response.status_code == 200 + rows = _rows_by_ticker(response.data) + assert rows['H2O'] == ['H2O', 10.0, 0.0, 10.0] diff --git a/backend/tests/analytics/api/test_analytics_plan_aggregate_viewset.py b/backend/tests/analytics/api/test_analytics_plan_aggregate_viewset.py new file mode 100644 index 0000000..bd046d4 --- /dev/null +++ b/backend/tests/analytics/api/test_analytics_plan_aggregate_viewset.py @@ -0,0 +1,120 @@ +from collections.abc import Callable + +import pytest +from analytics.models import AnalyticsPlanAggregate +from django.urls import reverse +from gamedata.models.game_planet import GamePlanet +from rest_framework.test import APIClient +from user.models import User + +pytestmark = pytest.mark.django_db + + +def _detail_url(planet_natural_id: str) -> str: + return reverse('analytics:planet-insight-detail', kwargs={'planet_natural_id': planet_natural_id}) + + +class TestAnalyticsPlanAggregateViewSetAccess: + def test_unauthenticated_access_is_allowed( + self, api_client: APIClient, planet_factory: Callable[..., GamePlanet] + ) -> None: + planet = planet_factory(planet_natural_id='OT-580b') + + response = api_client.get(_detail_url(planet.planet_natural_id)) + + assert response.status_code == 200 + + def test_authenticated_access_also_succeeds( + self, + api_client: APIClient, + planet_factory: Callable[..., GamePlanet], + user_factory: Callable[..., User], + ) -> None: + planet = planet_factory(planet_natural_id='OT-580b') + user = user_factory() + + response = api_client.as_user(user).get(_detail_url(planet.planet_natural_id)) # ty:ignore[unresolved-attribute] + + assert response.status_code == 200 + + def test_disallowed_http_method_returns_405( + self, api_client: APIClient, planet_factory: Callable[..., GamePlanet] + ) -> None: + planet = planet_factory(planet_natural_id='OT-580b') + + response = api_client.post(_detail_url(planet.planet_natural_id), data={}) + + assert response.status_code == 405 + + +class TestAnalyticsPlanAggregateViewSetRetrieve: + def test_retrieve_returns_serialized_aggregate( + self, + api_client: APIClient, + planet_factory: Callable[..., GamePlanet], + plan_aggregate_factory: Callable[..., AnalyticsPlanAggregate], + ) -> None: + planet = planet_factory(planet_natural_id='OT-580b') + aggregate = plan_aggregate_factory( + planet_natural_id=planet.planet_natural_id, + total_plans_analyzed=42, + insights_data={'avg_cost': 1234.5, 'materials': ['H2O', 'DW']}, + ) + + response = api_client.get(_detail_url(planet.planet_natural_id)) + + assert response.status_code == 200 + assert response.data['status'] == 'success' + assert response.data['planet_natural_id'] == planet.planet_natural_id + assert response.data['total_plans_analyzed'] == 42 + assert response.data['insights_data'] == {'avg_cost': 1234.5, 'materials': ['H2O', 'DW']} + assert response.data['last_updated'] is not None + assert response['X-Cache-Hit'] == '0' + assert aggregate.pk is not None + + def test_retrieve_planet_without_aggregate_returns_below_threshold( + self, api_client: APIClient, planet_factory: Callable[..., GamePlanet] + ) -> None: + planet = planet_factory(planet_natural_id='OT-580b') + + response = api_client.get(_detail_url(planet.planet_natural_id)) + + assert response.status_code == 200 + assert response.data == { + 'status': 'below_threshold', + 'planet_natural_id': planet.planet_natural_id, + 'total_plans_analyzed': 0, + 'aggregated_data': None, + } + + def test_retrieve_unknown_planet_returns_404(self, api_client: APIClient) -> None: + response = api_client.get(_detail_url('XX-000x')) + + assert response.status_code == 404 + assert response.data['detail'] == 'Planet not found.' + + def test_retrieve_only_returns_data_for_requested_planet( + self, + api_client: APIClient, + planet_factory: Callable[..., GamePlanet], + plan_aggregate_factory: Callable[..., AnalyticsPlanAggregate], + ) -> None: + planet_a = planet_factory(planet_natural_id='OT-580b') + planet_b = planet_factory(planet_natural_id='ZV-759b') + + plan_aggregate_factory(planet_natural_id=planet_a.planet_natural_id, total_plans_analyzed=5) + plan_aggregate_factory(planet_natural_id=planet_b.planet_natural_id, total_plans_analyzed=99) + + response_a = api_client.get(_detail_url(planet_a.planet_natural_id)) + response_b = api_client.get(_detail_url(planet_b.planet_natural_id)) + + assert response_a.data['total_plans_analyzed'] == 5 + assert response_b.data['total_plans_analyzed'] == 99 + + def test_retrieve_empty_database_does_not_error(self, api_client: APIClient) -> None: + assert AnalyticsPlanAggregate.objects.count() == 0 + assert GamePlanet.objects.count() == 0 + + response = api_client.get(_detail_url('OT-580b')) + + assert response.status_code == 404 diff --git a/backend/tests/gamedata/api/conftest.py b/backend/tests/gamedata/api/conftest.py new file mode 100644 index 0000000..e7c8bbc --- /dev/null +++ b/backend/tests/gamedata/api/conftest.py @@ -0,0 +1,16 @@ +from collections.abc import Callable + +import pytest +from gamedata.models.game_exchange import GameExchange +from model_bakery import baker +from user.models.configs import GlobalConfigWebhook + + +@pytest.fixture() +def exchange_factory() -> Callable[..., GameExchange]: + return lambda **kwargs: baker.make('gamedata.GameExchange', **kwargs) + + +@pytest.fixture() +def webhook_config_factory() -> Callable[..., GlobalConfigWebhook]: + return lambda **kwargs: baker.make('user.GlobalConfigWebhook', **kwargs) diff --git a/backend/tests/gamedata/api/test_gamedata_viewsets.py b/backend/tests/gamedata/api/test_gamedata_viewsets.py new file mode 100644 index 0000000..fc95769 --- /dev/null +++ b/backend/tests/gamedata/api/test_gamedata_viewsets.py @@ -0,0 +1,162 @@ +import uuid +from collections.abc import Callable +from datetime import timedelta +from unittest.mock import patch + +import pytest +from django.urls import reverse +from django.utils import timezone +from gamedata.models.game_exchange import GameExchange +from rest_framework.test import APIClient +from user.models.configs import GlobalConfigWebhook, WebhookSenderChoices + +pytestmark = pytest.mark.django_db + + +def _exchange_list_url() -> str: + return reverse('data:exchange-list') + + +def _exchange_csv_url() -> str: + return reverse('data:exchanges-list-csv') + + +def _webhook_url(token: uuid.UUID) -> str: + return reverse('data:fio-webhook-ingest', kwargs={'token': token}) + + +class TestGameExchangeViewSet: + def test_list_annotates_status_and_merges_live_data( + self, + api_client: APIClient, + exchange_analytics_factory: Callable[..., object], + exchange_factory: Callable[..., GameExchange], + ) -> None: + now = timezone.now().date() + stale_date = now - timedelta(days=5) + + exchange_analytics_factory( + ticker='FUEL', exchange_code='AI1', calendar_date=stale_date, vwap_7d=100, avg_traded_7d=10 + ) + exchange_analytics_factory(ticker='IRON', exchange_code='NC1', calendar_date=now, vwap_7d=50, avg_traded_7d=5) + exchange_analytics_factory(ticker='GOLD', exchange_code='CI1', calendar_date=now, vwap_7d=0, avg_traded_7d=0) + # exchange code not part of the tracked target list, must be excluded + exchange_analytics_factory(ticker='VOID', exchange_code='BAD_EXC') + + exchange_factory( + ticker_id='IRONNC1', ticker='IRON', exchange_code='NC1', ask=12.5, bid=11.5, supply=5, demand=42 + ) + + response = api_client.get(_exchange_list_url()) + + assert response.status_code == 200 + data = response.data + assert len(data) == 3 + + results = {item['ticker']: item for item in data} + + assert results['FUEL']['exchange_status'] == 'STALE' + assert results['IRON']['exchange_status'] == 'ACTIVE' + assert results['GOLD']['exchange_status'] == 'INACTIVE' + assert results['IRON']['ticker_id'] == 'IRON.NC1' + + # live data merged in from GameExchange + assert results['IRON']['ask'] == 12.5 + assert results['IRON']['bid'] == 11.5 + assert results['IRON']['supply'] == 5 + assert results['IRON']['demand'] == 42 + + # no live data available defaults to 0.0 + assert results['FUEL']['ask'] == 0.0 + assert results['FUEL']['bid'] == 0.0 + + def test_list_returns_duplicate_ticker_rows_on_sqlite( + self, api_client: APIClient, exchange_analytics_factory: Callable[..., object] + ) -> None: + exchange_analytics_factory(ticker='H2O', exchange_code='AI1', date_epoch=1000) + exchange_analytics_factory(ticker='H2O', exchange_code='AI1', date_epoch=5000) + + response = api_client.get(_exchange_list_url()) + + h2o_records = [row for row in response.data if row['ticker'] == 'H2O'] + + assert len(h2o_records) == 2 + assert h2o_records[0]['date_epoch'] == 5000 + assert h2o_records[1]['date_epoch'] == 1000 + + def test_list_empty_database_returns_empty_list(self, api_client: APIClient) -> None: + response = api_client.get(_exchange_list_url()) + + assert response.status_code == 200 + assert response.data == [] + + def test_unauthenticated_access_is_allowed(self, api_client: APIClient) -> None: + response = api_client.get(_exchange_list_url()) + + assert response.status_code == 200 + + +class TestGameExchangeCSVViewSet: + def test_csv_export_format_and_headers( + self, api_client: APIClient, exchange_analytics_factory: Callable[..., object] + ) -> None: + exchange_analytics_factory(ticker='FUEL', exchange_code='AI1', date_epoch=12345) + + response = api_client.get(_exchange_csv_url()) + + assert response.status_code == 200 + assert response['Content-Type'] == 'text/csv; charset=utf-8' + + content = response.content.decode('utf-8') + lines = content.splitlines() + + expected_header = ( + 'ticker,exchange_code,ticker_id,date_epoch,calendar_date,exchange_status,' + 'vwap_daily,vwap_7d,vwap_30d,traded_daily,sum_traded_7d,sum_traded_30d,' + 'avg_traded_7d,avg_traded_30d,ask,bid,supply,demand' + ) + + assert lines[0] == expected_header + assert 'FUEL,AI1' in lines[1] + + +class TestFIOWebhookIngest: + def test_unknown_token_returns_404(self, api_client: APIClient) -> None: + response = api_client.post(_webhook_url(uuid.uuid4()), data={'Data': []}, format='json') + + assert response.status_code == 404 + + def test_inactive_config_returns_404( + self, api_client: APIClient, webhook_config_factory: Callable[..., GlobalConfigWebhook] + ) -> None: + config = webhook_config_factory(sender=WebhookSenderChoices.FIOAPI, is_active=False) + + response = api_client.post(_webhook_url(config.path), data={'Data': []}, format='json') + + assert response.status_code == 404 + + def test_invalid_payload_returns_400( + self, api_client: APIClient, webhook_config_factory: Callable[..., GlobalConfigWebhook] + ) -> None: + config = webhook_config_factory(sender=WebhookSenderChoices.FIOAPI, is_active=True) + + response = api_client.post(_webhook_url(config.path), data={}, format='json') + + assert response.status_code == 400 + + def test_valid_payload_accepted_updates_stats_and_queues_task( + self, api_client: APIClient, webhook_config_factory: Callable[..., GlobalConfigWebhook] + ) -> None: + config = webhook_config_factory( + sender=WebhookSenderChoices.FIOAPI, is_active=True, total_calls=3, last_received_at=None + ) + + with patch('gamedata.api.viewsets.gamedata_process_fio_webhook.delay') as mock_delay: + response = api_client.post(_webhook_url(config.path), data={'Data': []}, format='json') + + assert response.status_code == 202 + mock_delay.assert_called_once_with({'Data': []}) + + config.refresh_from_db() + assert config.total_calls == 4 + assert config.last_received_at is not None diff --git a/backend/tests/planning/api/viewsets/test_cx_viewset.py b/backend/tests/planning/api/viewsets/test_cx_viewset.py new file mode 100644 index 0000000..e629c96 --- /dev/null +++ b/backend/tests/planning/api/viewsets/test_cx_viewset.py @@ -0,0 +1,141 @@ +from unittest.mock import patch + +import pytest +from django.urls import reverse + +pytestmark = pytest.mark.django_db + + +class TestCXViewSetCrud: + def test_list_requires_auth(self, api_client, user_factory, cx_factory): + url = reverse('planning:cx') + + response_noauth = api_client.get(url) + assert response_noauth.status_code == 401 + + user = user_factory(id=1) + cx_factory(user=user, cx_name='My CX') + + response = api_client.as_user(user).get(url) + assert response.status_code == 200 + assert len(response.data) == 1 + assert response.data[0]['cx_name'] == 'My CX' + + def test_retrieve_404_and_200(self, api_client, user_factory, cx_factory): + user = user_factory(id=1) + cx = cx_factory(user=user, cx_name='My CX') + + url_404 = reverse('planning:cx-detail', kwargs={'pk': '356da85a-494a-45a9-b20e-16d0f128c5b8'}) + response_404 = api_client.as_user(user).get(url_404) + assert response_404.status_code == 404 + + url = reverse('planning:cx-detail', kwargs={'pk': str(cx.uuid)}) + response = api_client.as_user(user).get(url) + assert response.status_code == 200 + assert response.data['uuid'] == str(cx.uuid) + + def test_create(self, api_client, user_factory): + user = user_factory(id=1) + url = reverse('planning:cx') + post_data = {'cx_name': 'New CX', 'cx_data': {}} + + response_noauth = api_client.post(url, data=post_data, format='json') + assert response_noauth.status_code == 401 + + response = api_client.as_user(user).post(url, data=post_data, format='json') + assert response.status_code == 201 + assert response.data['cx_name'] == 'New CX' + + def test_update(self, api_client, user_factory, cx_factory): + user = user_factory(id=1) + cx = cx_factory(user=user, cx_name='Old Name', cx_data={}) + + url = reverse('planning:cx-detail', kwargs={'pk': str(cx.uuid)}) + response = api_client.as_user(user).put(url, data={'cx_name': 'Updated Name', 'cx_data': {}}, format='json') + + assert response.status_code == 200 + assert response.data['cx_name'] == 'Updated Name' + + def test_destroy(self, api_client, user_factory, cx_factory): + user = user_factory(id=1) + cx = cx_factory(user=user, cx_name='To Delete') + + url = reverse('planning:cx-detail', kwargs={'pk': str(cx.uuid)}) + + response_noauth = api_client.delete(url) + assert response_noauth.status_code == 401 + + response = api_client.as_user(user).delete(url) + assert response.status_code == 204 + + +class TestCXViewSetSyncJunctions: + def test_sync_junctions_requires_auth(self, api_client): + url = reverse('planning:cx-junctions') + + response = api_client.post(url, data=[], format='json') + assert response.status_code == 401 + + def test_sync_junctions_assigns_empire_to_cx(self, api_client, user_factory, cx_factory, empire_factory): + user = user_factory(id=1) + cx = cx_factory(user=user, cx_name='My CX') + empire = empire_factory(user=user) + + url = reverse('planning:cx-junctions') + payload = [{'cx_uuid': str(cx.uuid), 'empires': [{'empire_uuid': str(empire.uuid)}]}] + + with patch('planning.api.viewsets.cx_viewset.PlanningCacheManager.delete_pattern') as mock_delete_pattern: + response = api_client.as_user(user).post(url, data=payload, format='json') + + assert response.status_code == 200 + mock_delete_pattern.assert_called_once_with(f'*PLANNING:{user.id}:*') + + empire.refresh_from_db() + assert empire.cx_id == cx.uuid + + def test_sync_junctions_rejects_unowned_cx(self, api_client, user_factory, cx_factory, empire_factory): + user = user_factory(id=1) + other_user = user_factory(id=2) + other_cx = cx_factory(user=other_user) + empire = empire_factory(user=user) + + url = reverse('planning:cx-junctions') + payload = [{'cx_uuid': str(other_cx.uuid), 'empires': [{'empire_uuid': str(empire.uuid)}]}] + + response = api_client.as_user(user).post(url, data=payload, format='json') + + assert response.status_code == 403 + assert response.data['error'] == 'Invalid CX UUIDs detected.' + + def test_sync_junctions_rejects_unowned_empire(self, api_client, user_factory, cx_factory, empire_factory): + user = user_factory(id=1) + other_user = user_factory(id=2) + cx = cx_factory(user=user) + other_empire = empire_factory(user=other_user) + + url = reverse('planning:cx-junctions') + payload = [{'cx_uuid': str(cx.uuid), 'empires': [{'empire_uuid': str(other_empire.uuid)}]}] + + response = api_client.as_user(user).post(url, data=payload, format='json') + + assert response.status_code == 403 + assert response.data['error'] == 'Invalid Empire UUIDs detected.' + + def test_sync_junctions_rejects_duplicate_empire_assignment( + self, api_client, user_factory, cx_factory, empire_factory + ): + user = user_factory(id=1) + cx_1 = cx_factory(user=user) + cx_2 = cx_factory(user=user) + empire = empire_factory(user=user) + + url = reverse('planning:cx-junctions') + payload = [ + {'cx_uuid': str(cx_1.uuid), 'empires': [{'empire_uuid': str(empire.uuid)}]}, + {'cx_uuid': str(cx_2.uuid), 'empires': [{'empire_uuid': str(empire.uuid)}]}, + ] + + response = api_client.as_user(user).post(url, data=payload, format='json') + + assert response.status_code == 400 + assert response.data['error'] == 'Duplicate empire assignment in request.' diff --git a/backend/tests/planning/api/viewsets/test_empire_viewset.py b/backend/tests/planning/api/viewsets/test_empire_viewset.py new file mode 100644 index 0000000..0cc4aae --- /dev/null +++ b/backend/tests/planning/api/viewsets/test_empire_viewset.py @@ -0,0 +1,216 @@ +from unittest.mock import patch + +import pytest +from django.urls import reverse +from planning.models import PlanningCOGCChoices, PlanningFactionChoices +from tests.fixtures.planning.fxt_plan_vallis import plan_data_vallis + +pytestmark = pytest.mark.django_db + + +def _empire_payload(**overrides): + payload = { + 'empire_name': 'My Empire', + 'empire_faction': PlanningFactionChoices.ANTARES, + 'empire_permits_used': 1, + 'empire_permits_total': 2, + } + payload.update(overrides) + return payload + + +class TestEmpireViewSetCrud: + def test_list_requires_auth(self, api_client, user_factory, empire_factory): + url = reverse('planning:empire') + + response_noauth = api_client.get(url) + assert response_noauth.status_code == 401 + + user = user_factory(id=1) + empire_factory(user=user, empire_name='My Empire') + + response = api_client.as_user(user).get(url) + assert response.status_code == 200 + assert len(response.data) == 1 + assert response.data[0]['empire_name'] == 'My Empire' + + def test_retrieve_404_and_200(self, api_client, user_factory, empire_factory): + user = user_factory(id=1) + empire = empire_factory(user=user, empire_name='My Empire') + + url_404 = reverse('planning:empire-detail', kwargs={'pk': '356da85a-494a-45a9-b20e-16d0f128c5b8'}) + response_404 = api_client.as_user(user).get(url_404) + assert response_404.status_code == 404 + + url = reverse('planning:empire-detail', kwargs={'pk': str(empire.uuid)}) + response = api_client.as_user(user).get(url) + assert response.status_code == 200 + assert response.data['uuid'] == str(empire.uuid) + + def test_retrieve_plans(self, api_client, user_factory, empire_factory, plan_factory): + user = user_factory(id=1) + empire = empire_factory(user=user) + plan = plan_factory(user=user, plan_data=plan_data_vallis) + empire.plans.add(plan, through_defaults={'user': user}) + + url = reverse('planning:empire-plan-list', kwargs={'pk': str(empire.uuid)}) + + response_noauth = api_client.get(url) + assert response_noauth.status_code == 401 + + response = api_client.as_user(user).get(url) + assert response.status_code == 200 + assert len(response.data) == 1 + assert response.data[0]['uuid'] == str(plan.uuid) + + def test_create(self, api_client, user_factory): + user = user_factory(id=1) + url = reverse('planning:empire') + + response_noauth = api_client.post(url, data=_empire_payload(), format='json') + assert response_noauth.status_code == 401 + + response = api_client.as_user(user).post(url, data=_empire_payload(), format='json') + assert response.status_code == 201 + assert response.data['empire_name'] == 'My Empire' + + def test_update(self, api_client, user_factory, empire_factory): + user = user_factory(id=1) + empire = empire_factory(user=user, empire_name='Old Name') + + url = reverse('planning:empire-detail', kwargs={'pk': str(empire.uuid)}) + response = api_client.as_user(user).put( + url, data=_empire_payload(empire_name='New Name'), format='json' + ) + + assert response.status_code == 200 + assert response.data['empire_name'] == 'New Name' + + def test_destroy(self, api_client, user_factory, empire_factory): + user = user_factory(id=1) + empire = empire_factory(user=user) + + url = reverse('planning:empire-detail', kwargs={'pk': str(empire.uuid)}) + + response_noauth = api_client.delete(url) + assert response_noauth.status_code == 401 + + response = api_client.as_user(user).delete(url) + assert response.status_code == 204 + + +class TestEmpireViewSetSyncJunctions: + def test_sync_junctions_requires_auth(self, api_client): + url = reverse('planning:empire-junctions') + + response = api_client.post(url, data=[], format='json') + assert response.status_code == 401 + + def test_sync_junctions_creates_and_removes_links( + self, api_client, user_factory, empire_factory, plan_factory + ): + user = user_factory(id=1) + empire = empire_factory(user=user) + plan_keep = plan_factory(user=user, plan_data=plan_data_vallis) + plan_new = plan_factory(user=user, plan_data=plan_data_vallis) + plan_drop = plan_factory(user=user, plan_data=plan_data_vallis) + + empire.plans.add(plan_keep, through_defaults={'user': user}) + empire.plans.add(plan_drop, through_defaults={'user': user}) + + url = reverse('planning:empire-junctions') + payload = [ + { + 'empire_uuid': str(empire.uuid), + 'baseplanners': [ + {'baseplanner_uuid': str(plan_keep.uuid)}, + {'baseplanner_uuid': str(plan_new.uuid)}, + ], + } + ] + + with patch('planning.api.viewsets.empire_viewset.PlanningCacheManager.delete_pattern') as mock_delete_pattern: + response = api_client.as_user(user).post(url, data=payload, format='json') + + assert response.status_code == 200 + mock_delete_pattern.assert_called_once_with(f'*PLANNING:{user.id}:*') + + linked_plan_uuids = set(empire.plans.values_list('uuid', flat=True)) + assert linked_plan_uuids == {plan_keep.uuid, plan_new.uuid} + + def test_sync_junctions_no_changes_skips_cache_invalidation( + self, api_client, user_factory, empire_factory, plan_factory + ): + user = user_factory(id=1) + empire = empire_factory(user=user) + plan = plan_factory(user=user, plan_data=plan_data_vallis) + empire.plans.add(plan, through_defaults={'user': user}) + + url = reverse('planning:empire-junctions') + payload = [{'empire_uuid': str(empire.uuid), 'baseplanners': [{'baseplanner_uuid': str(plan.uuid)}]}] + + with patch('planning.api.viewsets.empire_viewset.PlanningCacheManager.delete_pattern') as mock_delete_pattern: + response = api_client.as_user(user).post(url, data=payload, format='json') + + assert response.status_code == 200 + mock_delete_pattern.assert_not_called() + + def test_sync_junctions_rejects_unowned_references( + self, api_client, user_factory, empire_factory, plan_factory + ): + user = user_factory(id=1) + other_user = user_factory(id=2) + empire = empire_factory(user=user) + other_plan = plan_factory(user=other_user, plan_data=plan_data_vallis) + + url = reverse('planning:empire-junctions') + payload = [{'empire_uuid': str(empire.uuid), 'baseplanners': [{'baseplanner_uuid': str(other_plan.uuid)}]}] + + response = api_client.as_user(user).post(url, data=payload, format='json') + + assert response.status_code == 403 + assert response.data['invalid_plans'] == [other_plan.uuid] + + +class TestEmpireViewSetSyncState: + def test_sync_state_requires_auth(self, api_client, user_factory, empire_factory): + user = user_factory(id=1) + empire = empire_factory(user=user) + + url = reverse('planning:empire-sync-state', kwargs={'pk': str(empire.uuid)}) + response = api_client.patch(url, data={}, format='json') + + assert response.status_code == 401 + + def test_sync_state_updates_empire_state(self, api_client, user_factory, empire_factory, plan_factory): + user = user_factory(id=1) + empire = empire_factory(user=user, empire_state={}, needs_state_sync=False) + plan = plan_factory(user=user, plan_data=plan_data_vallis, planet_natural_id='OT-580b') + + url = reverse('planning:empire-sync-state', kwargs={'pk': str(empire.uuid)}) + payload = { + 'metadata': { + 'faction': PlanningFactionChoices.ANTARES, + 'permits_used': 1, + 'permits_total': 2, + 'plan_count': 1, + 'timestamp': '2026-01-01T00:00:00Z', + }, + 'empire_total': {'H2O': {'p': 10.0, 'c': 5.0, 'd': 5.0}}, + 'plan_details': { + str(plan.uuid): { + 'metadata': {'planet_natural_id': 'OT-580b', 'cogc': PlanningCOGCChoices.NONE}, + 'deltas': {'H2O': {'p': 10.0, 'c': 5.0, 'd': 5.0}}, + } + }, + } + + with patch('planning.api.viewsets.empire_viewset.PlanningCacheManager.delete_pattern') as mock_delete_pattern: + response = api_client.as_user(user).patch(url, data=payload, format='json') + + assert response.status_code == 200 + mock_delete_pattern.assert_called_once_with(f'*PLANNING:{user.id}:*') + + empire.refresh_from_db() + assert empire.needs_state_sync is True + assert empire.empire_state['empire_total']['H2O']['p'] == 10.0 diff --git a/backend/tests/planning/api/viewsets/test_plan_viewset.py b/backend/tests/planning/api/viewsets/test_plan_viewset.py new file mode 100644 index 0000000..0df97ca --- /dev/null +++ b/backend/tests/planning/api/viewsets/test_plan_viewset.py @@ -0,0 +1,128 @@ +import pytest +from django.urls import reverse +from planning.models import PlanningPlan +from tests.fixtures.planning.fxt_plan_vallis import plan_data_vallis + +pytestmark = pytest.mark.django_db + + +def _plan_payload(**overrides): + payload = { + 'plan_name': 'My Plan', + 'planet_natural_id': 'OT-580b', + 'plan_permits_used': 1, + 'plan_corphq': False, + 'plan_data': plan_data_vallis, + } + payload.update(overrides) + return payload + + +class TestPlanViewSetCrud: + def test_list_requires_auth(self, api_client, user_factory, plan_factory): + url = reverse('planning:plan') + + response_noauth = api_client.get(url) + assert response_noauth.status_code == 401 + + user = user_factory(id=1) + plan_factory(user=user, plan_data=plan_data_vallis, plan_name='My Plan') + + response = api_client.as_user(user).get(url) + assert response.status_code == 200 + assert len(response.data) == 1 + assert response.data[0]['plan_name'] == 'My Plan' + + def test_retrieve_404_and_200(self, api_client, user_factory, plan_factory): + user = user_factory(id=1) + plan = plan_factory(user=user, plan_data=plan_data_vallis) + + url_404 = reverse('planning:plan-detail', kwargs={'pk': '356da85a-494a-45a9-b20e-16d0f128c5b8'}) + response_404 = api_client.as_user(user).get(url_404) + assert response_404.status_code == 404 + + url = reverse('planning:plan-detail', kwargs={'pk': str(plan.uuid)}) + response = api_client.as_user(user).get(url) + assert response.status_code == 200 + assert response.data['uuid'] == str(plan.uuid) + + def test_retrieve_only_returns_own_plans(self, api_client, user_factory, plan_factory): + user = user_factory(id=1) + other_user = user_factory(id=2) + other_plan = plan_factory(user=other_user, plan_data=plan_data_vallis) + + url = reverse('planning:plan-detail', kwargs={'pk': str(other_plan.uuid)}) + response = api_client.as_user(user).get(url) + + assert response.status_code == 404 + + def test_create(self, api_client, user_factory): + user = user_factory(id=1) + url = reverse('planning:plan') + + response_noauth = api_client.post(url, data=_plan_payload(), format='json') + assert response_noauth.status_code == 401 + + response = api_client.as_user(user).post(url, data=_plan_payload(), format='json') + assert response.status_code == 201 + assert response.data['plan_name'] == 'My Plan' + + def test_create_links_empire_when_empire_uuid_provided(self, api_client, user_factory, empire_factory): + user = user_factory(id=1) + empire = empire_factory(user=user) + + url = reverse('planning:plan') + response = api_client.as_user(user).post( + url, data=_plan_payload(empire_uuid=str(empire.uuid)), format='json' + ) + + assert response.status_code == 201 + empire.refresh_from_db() + assert empire.plans.count() == 1 + + def test_update(self, api_client, user_factory, plan_factory): + user = user_factory(id=1) + plan = plan_factory(user=user, plan_data=plan_data_vallis, plan_name='Old Name') + + url = reverse('planning:plan-detail', kwargs={'pk': str(plan.uuid)}) + response = api_client.as_user(user).put( + url, data=_plan_payload(plan_name='New Name'), format='json' + ) + + assert response.status_code == 200 + assert response.data['plan_name'] == 'New Name' + + def test_destroy(self, api_client, user_factory, plan_factory): + user = user_factory(id=1) + plan = plan_factory(user=user, plan_data=plan_data_vallis) + + url = reverse('planning:plan-detail', kwargs={'pk': str(plan.uuid)}) + + response_noauth = api_client.delete(url) + assert response_noauth.status_code == 401 + + response = api_client.as_user(user).delete(url) + assert response.status_code == 204 + + +class TestPlanViewSetClone: + def test_clone_requires_auth(self, api_client, user_factory, plan_factory): + user = user_factory(id=1) + plan = plan_factory(user=user, plan_data=plan_data_vallis) + + url = reverse('planning:plan-clone', kwargs={'pk': str(plan.uuid)}) + response = api_client.post(url) + + assert response.status_code == 401 + + def test_clone_creates_new_plan_with_suffixed_name(self, api_client, user_factory, plan_factory): + user = user_factory(id=1) + plan = plan_factory(user=user, plan_data=plan_data_vallis, plan_name='Original') + + url = reverse('planning:plan-clone', kwargs={'pk': str(plan.uuid)}) + response = api_client.as_user(user).post(url) + + assert response.status_code == 201 + assert response.data['plan_name'] == 'Original (Clone)' + assert response.data['uuid'] != str(plan.uuid) + assert PlanningPlan.objects.filter(user=user).count() == 2 diff --git a/backend/tests/planning/conftest.py b/backend/tests/planning/conftest.py index c7c45ad..71686c7 100644 --- a/backend/tests/planning/conftest.py +++ b/backend/tests/planning/conftest.py @@ -10,3 +10,13 @@ def plan_factory(**kwargs): @pytest.fixture() def shared_factory(**kwargs): return lambda **kwargs: baker.make('planning.PlanningShared', **kwargs) + + +@pytest.fixture() +def cx_factory(**kwargs): + return lambda **kwargs: baker.make('planning.PlanningCX', **kwargs) + + +@pytest.fixture() +def empire_factory(**kwargs): + return lambda **kwargs: baker.make('planning.PlanningEmpire', **kwargs) diff --git a/backend/tests/user/api/test_user_viewsets.py b/backend/tests/user/api/test_user_viewsets.py new file mode 100644 index 0000000..a212d8e --- /dev/null +++ b/backend/tests/user/api/test_user_viewsets.py @@ -0,0 +1,331 @@ +from unittest.mock import patch + +import pytest +from django.urls import reverse +from model_bakery import baker +from planning.models import PlanningCX, PlanningEmpire +from rest_framework_simplejwt.tokens import RefreshToken +from user.api.serializer import UserChangePasswordSerializer, UserProfileSerializer +from user.api.viewsets import UserProfileViewSet +from user.models import User, UserAPIKey, UserPreference +from user.models.verification_codes import VerificationCode, VerificationeCodeChoices + +pytestmark = pytest.mark.django_db + + +class TestUserPreferenceViewSet: + def test_retrieve_requires_auth(self, api_client): + response = api_client.get(reverse('user:user_preferences')) + assert response.status_code == 401 + + def test_retrieve_returns_defaults_when_unset(self, api_client, user_factory): + user = user_factory(id=1) + + response = api_client.as_user(user).get(reverse('user:user_preferences')) + + assert response.status_code == 200 + assert response.data['locale'] == 'en_US' + assert response.data['burnDaysRed'] == 5 + assert UserPreference.objects.filter(user=user).exists() + + def test_update_persists_preferences(self, api_client, user_factory): + user = user_factory(id=1) + + response = api_client.as_user(user).patch( + reverse('user:user_preferences'), data={'locale': 'de_DE', 'burnDaysRed': 3}, format='json' + ) + + assert response.status_code == 200 + assert response.data['locale'] == 'de_DE' + assert response.data['burnDaysRed'] == 3 + + preference = UserPreference.objects.get(user=user) + assert preference.preferences['locale'] == 'de_DE' + + +class TestUserRegisterViewSet: + def _payload(self, **overrides): + payload = { + 'username': 'newpilot', + 'password': 'Xk7!qzR9pLm2', + 'email': 'newpilot@example.com', + 'planet_id': 'OT-580b', + 'planet_input': 'montem', + } + payload.update(overrides) + return payload + + def test_register_creates_user_cx_and_empire(self, api_client): + with patch('user.tasks.send_email_verification_code.apply_async'): + response = api_client.post(reverse('user:user_signup'), data=self._payload(), format='json') + + assert response.status_code == 201 + assert response.data['username'] == 'newpilot' + + user = User.objects.get(username='newpilot') + assert PlanningCX.objects.filter(user=user).exists() + assert PlanningEmpire.objects.filter(user=user).exists() + + def test_register_rejects_wrong_planet_captcha(self, api_client): + response = api_client.post( + reverse('user:user_signup'), data=self._payload(planet_input='wrong'), format='json' + ) + assert response.status_code == 400 + + def test_register_rejects_duplicate_username(self, api_client, user_factory): + user_factory(username='newpilot') + + response = api_client.post(reverse('user:user_signup'), data=self._payload(), format='json') + assert response.status_code == 400 + + +class TestUserAPIKeyViewSet: + def test_list_requires_auth(self, api_client): + response = api_client.get(reverse('user:user_apikey_list')) + assert response.status_code == 401 + + def test_create_returns_key_material_once(self, api_client, user_factory): + user = user_factory(id=1) + + response = api_client.as_user(user).post( + reverse('user:user_apikey_list'), data={'name': 'my key'}, format='json' + ) + + assert response.status_code == 201 + assert response.data['name'] == 'my key' + assert 'api_key' in response.data and response.data['api_key'] + + def test_list_only_returns_own_keys(self, api_client, user_factory): + user = user_factory(id=1) + other_user = user_factory(id=2) + + UserAPIKey.objects.create_key(name='mine', user=user) + UserAPIKey.objects.create_key(name='theirs', user=other_user) + + response = api_client.as_user(user).get(reverse('user:user_apikey_list')) + + assert response.status_code == 200 + assert len(response.data) == 1 + assert response.data[0]['name'] == 'mine' + + def test_destroy_own_key(self, api_client, user_factory): + user = user_factory(id=1) + + api_key, _key = UserAPIKey.objects.create_key(name='mine', user=user) + + url = reverse('user:user_apikey_detail', kwargs={'pk': api_key.id}) + + response_noauth = api_client.delete(url) + assert response_noauth.status_code == 401 + + response = api_client.as_user(user).delete(url) + assert response.status_code == 204 + assert not UserAPIKey.objects.filter(id=api_key.id).exists() + + def test_destroy_other_users_key_returns_404(self, api_client, user_factory): + user = user_factory(id=1) + other_user = user_factory(id=2) + + api_key, _key = UserAPIKey.objects.create_key(name='theirs', user=other_user) + + url = reverse('user:user_apikey_detail', kwargs={'pk': api_key.id}) + response = api_client.as_user(user).delete(url) + + assert response.status_code == 404 + + +class TestUserEmailVerificationViewSet: + def test_request_code_requires_auth(self, api_client): + response = api_client.post(reverse('user:user_request_email_verification')) + assert response.status_code == 401 + + def test_request_code_already_verified_returns_400(self, api_client, user_factory): + user = user_factory(id=1, is_email_verified=True) + + response = api_client.as_user(user).post(reverse('user:user_request_email_verification')) + + assert response.status_code == 400 + + def test_request_code_sends_email_for_unverified_user(self, api_client, user_factory): + with patch('user.tasks.send_email_verification_code.apply_async') as mock_apply_async: + user = user_factory(id=1, is_email_verified=False, email='pilot@example.com') + mock_apply_async.reset_mock() + + response = api_client.as_user(user).post(reverse('user:user_request_email_verification')) + + assert response.status_code == 200 + mock_apply_async.assert_called_once() + assert VerificationCode.objects.filter( + user=user, purpose=VerificationeCodeChoices.EMAIL_VERIFICATION + ).exists() + + def test_verify_email_with_valid_code(self, api_client, user_factory): + user = user_factory(id=1, is_email_verified=False) + baker.make( + 'user.VerificationCode', user=user, code='ABCD1234', purpose=VerificationeCodeChoices.EMAIL_VERIFICATION + ) + + response = api_client.as_user(user).post( + reverse('user:user_verify_email'), data={'code': 'abcd1234'}, format='json' + ) + + assert response.status_code == 200 + user.refresh_from_db() + assert user.is_email_verified is True + + def test_verify_email_with_invalid_code_returns_400(self, api_client, user_factory): + user = user_factory(id=1, is_email_verified=False) + baker.make( + 'user.VerificationCode', user=user, code='ABCD1234', purpose=VerificationeCodeChoices.EMAIL_VERIFICATION + ) + + response = api_client.as_user(user).post( + reverse('user:user_verify_email'), data={'code': 'wrongcod'}, format='json' + ) + + assert response.status_code == 400 + + def test_verify_email_with_malformed_code_returns_serializer_errors(self, api_client, user_factory): + user = user_factory(id=1) + + response = api_client.as_user(user).post( + reverse('user:user_verify_email'), data={'code': 'short'}, format='json' + ) + + assert response.status_code == 400 + assert 'code' in response.data + + +class TestCustomTokenRefreshView: + def test_refresh_with_valid_token_queues_post_refresh_task(self, api_client, user_factory): + user = user_factory(id=1) + refresh = RefreshToken.for_user(user) + + with patch('user.api.viewsets.user_handle_post_refresh.delay') as mock_delay: + response = api_client.post( + reverse('user:token_refresh'), data={'refresh': str(refresh)}, format='json' + ) + + assert response.status_code == 200 + assert 'access' in response.data + mock_delay.assert_called_once_with(str(user.id)) + + def test_refresh_with_invalid_token_returns_401(self, api_client): + with patch('user.api.viewsets.user_handle_post_refresh.delay') as mock_delay: + response = api_client.post( + reverse('user:token_refresh'), data={'refresh': 'not-a-real-token'}, format='json' + ) + + assert response.status_code == 401 + mock_delay.assert_not_called() + + +class TestUserPasswordResetViewSet: + def test_request_code_for_known_verified_user(self, api_client, user_factory): + user = user_factory(id=1, email='pilot@example.com', is_email_verified=True) + + with patch('user.tasks.send_password_reset_code.apply_async') as mock_apply_async: + response = api_client.post( + reverse('user:user_request_password_reset'), data={'email': user.email}, format='json' + ) + + assert response.status_code == 200 + mock_apply_async.assert_called_once() + + def test_request_code_for_unknown_email_does_not_send(self, api_client): + with patch('user.tasks.send_password_reset_code.apply_async') as mock_apply_async: + response = api_client.post( + reverse('user:user_request_password_reset'), data={'email': 'nobody@example.com'}, format='json' + ) + + assert response.status_code == 200 + mock_apply_async.assert_not_called() + + def test_password_reset_with_valid_code(self, api_client, user_factory): + user = user_factory(id=1, email='pilot@example.com', is_email_verified=True) + baker.make( + 'user.VerificationCode', user=user, code='RESET123', purpose=VerificationeCodeChoices.PASSWORD_RESET + ) + + response = api_client.post( + reverse('user:user_password_reset'), + data={'email': user.email, 'code': 'RESET123', 'new_password': 'Xk7!qzR9pLm2'}, + format='json', + ) + + assert response.status_code == 200 + user.refresh_from_db() + assert user.check_password('Xk7!qzR9pLm2') + + def test_password_reset_with_invalid_code_returns_400(self, api_client, user_factory): + user = user_factory(id=1, email='pilot@example.com', is_email_verified=True) + + response = api_client.post( + reverse('user:user_password_reset'), + data={'email': user.email, 'code': 'WRONGCOD', 'new_password': 'Xk7!qzR9pLm2'}, + format='json', + ) + + assert response.status_code == 400 + + +class TestUserProfileViewSet: + def test_retrieve_requires_auth(self, api_client): + response = api_client.get(reverse('user:user_profile')) + assert response.status_code == 401 + + def test_retrieve_returns_profile(self, api_client, user_factory): + user = user_factory(id=1, username='pilot') + + response = api_client.as_user(user).get(reverse('user:user_profile')) + + assert response.status_code == 200 + assert response.data['username'] == 'pilot' + + def test_update_profile(self, api_client, user_factory): + user = user_factory(id=1) + + response = api_client.as_user(user).patch( + reverse('user:user_profile'), data={'prun_username': 'PilotName'}, format='json' + ) + + assert response.status_code == 200 + user.refresh_from_db() + assert user.prun_username == 'PilotName' + + def test_change_password_wrong_old_password_returns_400(self, api_client, user_factory): + user = user_factory(id=1) + user.set_password('CorrectHorse1!') + user.save() + + response = api_client.as_user(user).post( + reverse('user:user_change_password'), + data={'old_password': 'WrongPassword', 'new_password': 'Xk7!qzR9pLm2'}, + format='json', + ) + + assert response.status_code == 400 + + def test_change_password_success(self, api_client, user_factory): + user = user_factory(id=1) + user.set_password('CorrectHorse1!') + user.save() + + response = api_client.as_user(user).post( + reverse('user:user_change_password'), + data={'old_password': 'CorrectHorse1!', 'new_password': 'Xk7!qzR9pLm2'}, + format='json', + ) + + assert response.status_code == 200 + user.refresh_from_db() + assert user.check_password('Xk7!qzR9pLm2') + + def test_get_serializer_class_depends_on_action(self): + viewset = UserProfileViewSet() + + viewset.action = 'change_password' + assert viewset.get_serializer_class() is UserChangePasswordSerializer + + viewset.action = 'retrieve' + assert viewset.get_serializer_class() is UserProfileSerializer From a9b6c86f72043af9cb8cd588217b0ecb0b64d786 Mon Sep 17 00:00:00 2001 From: Jan Date: Sat, 5 Sep 2026 19:16:09 +0200 Subject: [PATCH 2/2] improve(tests): ruff --- .../api/test_analytics_market_insight_viewset.py | 16 ++++------------ .../planning/api/viewsets/test_empire_viewset.py | 12 +++--------- .../planning/api/viewsets/test_plan_viewset.py | 8 ++------ backend/tests/user/api/test_user_viewsets.py | 16 ++++------------ 4 files changed, 13 insertions(+), 39 deletions(-) diff --git a/backend/tests/analytics/api/test_analytics_market_insight_viewset.py b/backend/tests/analytics/api/test_analytics_market_insight_viewset.py index e08464f..a7ff60a 100644 --- a/backend/tests/analytics/api/test_analytics_market_insight_viewset.py +++ b/backend/tests/analytics/api/test_analytics_market_insight_viewset.py @@ -26,9 +26,7 @@ def test_unauthenticated_access_is_allowed(self, api_client: APIClient) -> None: assert response.status_code == 200 - def test_authenticated_access_also_succeeds( - self, api_client: APIClient, user_factory: Callable[..., User] - ) -> None: + def test_authenticated_access_also_succeeds(self, api_client: APIClient, user_factory: Callable[..., User]) -> None: user = user_factory() response = api_client.as_user(user).get(_url()) # ty:ignore[unresolved-attribute] @@ -59,15 +57,9 @@ def test_aggregates_production_consumption_and_delta_per_material( empire_a = empire_factory() empire_b = empire_factory() - material_snapshot_factory( - empire=empire_a, material_ticker='H2O', production=100, consumption=40, delta=60 - ) - material_snapshot_factory( - empire=empire_b, material_ticker='H2O', production=50, consumption=10, delta=40 - ) - material_snapshot_factory( - empire=empire_a, material_ticker='DW', production=5, consumption=5, delta=0 - ) + material_snapshot_factory(empire=empire_a, material_ticker='H2O', production=100, consumption=40, delta=60) + material_snapshot_factory(empire=empire_b, material_ticker='H2O', production=50, consumption=10, delta=40) + material_snapshot_factory(empire=empire_a, material_ticker='DW', production=5, consumption=5, delta=0) response = api_client.get(_url()) diff --git a/backend/tests/planning/api/viewsets/test_empire_viewset.py b/backend/tests/planning/api/viewsets/test_empire_viewset.py index 0cc4aae..1a11d75 100644 --- a/backend/tests/planning/api/viewsets/test_empire_viewset.py +++ b/backend/tests/planning/api/viewsets/test_empire_viewset.py @@ -79,9 +79,7 @@ def test_update(self, api_client, user_factory, empire_factory): empire = empire_factory(user=user, empire_name='Old Name') url = reverse('planning:empire-detail', kwargs={'pk': str(empire.uuid)}) - response = api_client.as_user(user).put( - url, data=_empire_payload(empire_name='New Name'), format='json' - ) + response = api_client.as_user(user).put(url, data=_empire_payload(empire_name='New Name'), format='json') assert response.status_code == 200 assert response.data['empire_name'] == 'New Name' @@ -106,9 +104,7 @@ def test_sync_junctions_requires_auth(self, api_client): response = api_client.post(url, data=[], format='json') assert response.status_code == 401 - def test_sync_junctions_creates_and_removes_links( - self, api_client, user_factory, empire_factory, plan_factory - ): + def test_sync_junctions_creates_and_removes_links(self, api_client, user_factory, empire_factory, plan_factory): user = user_factory(id=1) empire = empire_factory(user=user) plan_keep = plan_factory(user=user, plan_data=plan_data_vallis) @@ -155,9 +151,7 @@ def test_sync_junctions_no_changes_skips_cache_invalidation( assert response.status_code == 200 mock_delete_pattern.assert_not_called() - def test_sync_junctions_rejects_unowned_references( - self, api_client, user_factory, empire_factory, plan_factory - ): + def test_sync_junctions_rejects_unowned_references(self, api_client, user_factory, empire_factory, plan_factory): user = user_factory(id=1) other_user = user_factory(id=2) empire = empire_factory(user=user) diff --git a/backend/tests/planning/api/viewsets/test_plan_viewset.py b/backend/tests/planning/api/viewsets/test_plan_viewset.py index 0df97ca..7c7ce51 100644 --- a/backend/tests/planning/api/viewsets/test_plan_viewset.py +++ b/backend/tests/planning/api/viewsets/test_plan_viewset.py @@ -72,9 +72,7 @@ def test_create_links_empire_when_empire_uuid_provided(self, api_client, user_fa empire = empire_factory(user=user) url = reverse('planning:plan') - response = api_client.as_user(user).post( - url, data=_plan_payload(empire_uuid=str(empire.uuid)), format='json' - ) + response = api_client.as_user(user).post(url, data=_plan_payload(empire_uuid=str(empire.uuid)), format='json') assert response.status_code == 201 empire.refresh_from_db() @@ -85,9 +83,7 @@ def test_update(self, api_client, user_factory, plan_factory): plan = plan_factory(user=user, plan_data=plan_data_vallis, plan_name='Old Name') url = reverse('planning:plan-detail', kwargs={'pk': str(plan.uuid)}) - response = api_client.as_user(user).put( - url, data=_plan_payload(plan_name='New Name'), format='json' - ) + response = api_client.as_user(user).put(url, data=_plan_payload(plan_name='New Name'), format='json') assert response.status_code == 200 assert response.data['plan_name'] == 'New Name' diff --git a/backend/tests/user/api/test_user_viewsets.py b/backend/tests/user/api/test_user_viewsets.py index a212d8e..51d2dc7 100644 --- a/backend/tests/user/api/test_user_viewsets.py +++ b/backend/tests/user/api/test_user_viewsets.py @@ -67,9 +67,7 @@ def test_register_creates_user_cx_and_empire(self, api_client): assert PlanningEmpire.objects.filter(user=user).exists() def test_register_rejects_wrong_planet_captcha(self, api_client): - response = api_client.post( - reverse('user:user_signup'), data=self._payload(planet_input='wrong'), format='json' - ) + response = api_client.post(reverse('user:user_signup'), data=self._payload(planet_input='wrong'), format='json') assert response.status_code == 400 def test_register_rejects_duplicate_username(self, api_client, user_factory): @@ -155,9 +153,7 @@ def test_request_code_sends_email_for_unverified_user(self, api_client, user_fac assert response.status_code == 200 mock_apply_async.assert_called_once() - assert VerificationCode.objects.filter( - user=user, purpose=VerificationeCodeChoices.EMAIL_VERIFICATION - ).exists() + assert VerificationCode.objects.filter(user=user, purpose=VerificationeCodeChoices.EMAIL_VERIFICATION).exists() def test_verify_email_with_valid_code(self, api_client, user_factory): user = user_factory(id=1, is_email_verified=False) @@ -202,9 +198,7 @@ def test_refresh_with_valid_token_queues_post_refresh_task(self, api_client, use refresh = RefreshToken.for_user(user) with patch('user.api.viewsets.user_handle_post_refresh.delay') as mock_delay: - response = api_client.post( - reverse('user:token_refresh'), data={'refresh': str(refresh)}, format='json' - ) + response = api_client.post(reverse('user:token_refresh'), data={'refresh': str(refresh)}, format='json') assert response.status_code == 200 assert 'access' in response.data @@ -243,9 +237,7 @@ def test_request_code_for_unknown_email_does_not_send(self, api_client): def test_password_reset_with_valid_code(self, api_client, user_factory): user = user_factory(id=1, email='pilot@example.com', is_email_verified=True) - baker.make( - 'user.VerificationCode', user=user, code='RESET123', purpose=VerificationeCodeChoices.PASSWORD_RESET - ) + baker.make('user.VerificationCode', user=user, code='RESET123', purpose=VerificationeCodeChoices.PASSWORD_RESET) response = api_client.post( reverse('user:user_password_reset'),