improve(tests): increase coverage

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)

backend/tests/analytics/api/test_analytics_market_insight_viewset.py

@@ -0,0 +1,108 @@
+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]

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

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)