Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,11 @@ python_files = ["test_*.py"]
# "." resolves the `tests` suite package; "tests" resolves the test apps
# (test_project, excluded_app) that live inside it as top-level imports.
pythonpath = [".", "tests"]
# Treat an unused eager load as a hard failure by default:
# any UnusedRelatedAccess that escapes an explicit capture block fails the test.
filterwarnings = [
"error::eagle.exceptions.UnusedRelatedAccess",
]

[tool.coverage.run]
branch = true
Expand Down
93 changes: 40 additions & 53 deletions tests/base.py
Original file line number Diff line number Diff line change
@@ -1,89 +1,76 @@
import warnings
from types import SimpleNamespace
from dataclasses import dataclass

import pytest
from django.urls import reverse
from rest_framework.test import APIClient

from eagle import UnusedRelatedAccess, unused
from eagle import unused
from test_project.models import Climate, Eagle, Eaglet, Location
from tests.factories import ClimateFactory, EagleFactory, EagletFactory, LocationFactory


@dataclass(frozen=True)
class EagleGraph:
"""A connected set of rows: a location with one climate, an eagle living there, and its eaglet."""

location: Location
climate: Climate
eagle: Eagle
eaglet: Eaglet


@dataclass(frozen=True)
class EagleWithLocation:
"""An eagle paired with the location it lives in."""

location: Location
eagle: Eagle


@pytest.mark.django_db
class BaseRequestTest:
"""Base for tests that drive the eagle detail endpoint; an unused eager load surfaces as an error."""

@pytest.fixture
def api_client(self):
def api_client(self) -> APIClient:
"""Return a DRF test client for calling the eagle endpoints."""
return APIClient()

@pytest.fixture
def eagle_graph(self, db):
def eagle_graph(self, db: None) -> EagleGraph:
"""Build a fully connected eagle graph whose relations a request can eager-load."""
location = LocationFactory(climates=[ClimateFactory()])
eagle = EagleFactory(location=location)
eaglet = EagletFactory(eagle=eagle)
return SimpleNamespace(
return EagleGraph(
location=location,
climate=location.climates.get(),
eagle=eagle,
eaglet=eaglet,
)

@pytest.fixture
def warn_request(self, api_client, eagle_graph):
def _request(pk=None, **params):
target = eagle_graph.eagle.pk if pk is None else pk

url = reverse("eagle-detail", kwargs={"pk": target})

with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always", UnusedRelatedAccess)
response = api_client.get(url, params)

assert response.status_code == 200, response.content
return [w.message for w in caught if issubclass(w.category, UnusedRelatedAccess)]

return _request

@pytest.fixture(autouse=True)
def setup(self, warn_request, eagle_graph):
self.request = warn_request
self.graph = eagle_graph


@pytest.mark.django_db
class MayAccessHelperTestBase:
"""Base for tests of the may_access/mark_considered helpers run inside a tracking request."""

@pytest.fixture
def eagle_with_location(self, db):
def eagle_with_location(self, db: None) -> EagleWithLocation:
"""Create a single eagle and the location it lives in."""
location = LocationFactory()
eagle = EagleFactory(location=location)
return SimpleNamespace(location=location, eagle=eagle)

@pytest.fixture
def active_request(self):
unused.begin_request()
yield
if unused.is_active():
unused.end_request()

@pytest.fixture
def flush_warnings(self, active_request):
def _flush():
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always", UnusedRelatedAccess)
unused.end_request()
unused.begin_request()
return [w.message for w in caught if issubclass(w.category, UnusedRelatedAccess)]

return _flush
return EagleWithLocation(location=location, eagle=eagle)

@pytest.fixture(autouse=True)
def setup(self, eagle_with_location, flush_warnings):
self.data = eagle_with_location
self.flush = flush_warnings
def active_request(self) -> None:
"""Open a tracking request before each test; the global reset fixture closes it afterwards."""
unused.begin_request()


class InactiveCollectorTestBase:
"""Base for tests that must run with no active tracking request."""

@pytest.fixture(autouse=True)
def inactive_collector(self):
def inactive_collector(self) -> None:
"""Ensure no tracking request is active before the test body runs."""
if unused.is_active():
unused.end_request()
return
22 changes: 22 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
import warnings
from collections.abc import Iterator

import pytest

from eagle import UnusedRelatedAccess, unused


@pytest.fixture(autouse=True)
def reset_eagle_tracking() -> Iterator[None]:
"""
Force Eagle's per-request tracking inactive after every test so state never leaks between tests.

Warnings are configured as errors (see ``filterwarnings`` in pyproject.toml), so a request that
ends with an unused relation raises ``UnusedRelatedAccess`` part-way through ``end_request`` and
never reaches its own cleanup. This finalizer ignores that warning and resets the collector.
"""
yield
if unused.is_active():
with warnings.catch_warnings():
warnings.simplefilter("ignore", UnusedRelatedAccess)
unused.end_request()
20 changes: 14 additions & 6 deletions tests/test_enabled.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,11 @@
import pytest
from django.test import override_settings
from django.urls import reverse
from rest_framework.test import APIClient

from eagle import UnusedRelatedAccess
from eagle.config import is_enabled
from tests.base import BaseRequestTest
from tests.base import BaseRequestTest, EagleGraph


class TestEnabledDefault:
Expand All @@ -20,8 +24,12 @@ def test_enabled_when_setting_true(self):

class TestDisabledRequest(BaseRequestTest):
@override_settings(EAGLE_ENABLED=False)
def test_unused_eager_load_does_not_warn_when_disabled(self):
assert self.request(select_related="location") == []

def test_unused_eager_load_warns_when_enabled(self):
assert len(self.request(select_related="location")) == 1
def test_unused_eager_load_does_not_warn_when_disabled(self, api_client: APIClient, eagle_graph: EagleGraph):
url = reverse("eagle-detail", kwargs={"pk": eagle_graph.eagle.pk})
response = api_client.get(url, {"select_related": "location"})
assert response.status_code == 200

def test_unused_eager_load_warns_when_enabled(self, api_client: APIClient, eagle_graph: EagleGraph):
url = reverse("eagle-detail", kwargs={"pk": eagle_graph.eagle.pk})
with pytest.raises(UnusedRelatedAccess):
api_client.get(url, {"select_related": "location"})
47 changes: 31 additions & 16 deletions tests/test_mark_considered.py
Original file line number Diff line number Diff line change
@@ -1,23 +1,38 @@
from tests.base import BaseRequestTest
import pytest
from django.urls import reverse
from rest_framework.test import APIClient

from eagle import UnusedRelatedAccess
from tests.base import BaseRequestTest, EagleGraph


class TestMarkConsidered(BaseRequestTest):
def test_mark_considered_suppresses_warning(self):
assert self.request(select_related="location", mark_considered="location") == []
def test_mark_considered_suppresses_warning(self, api_client: APIClient, eagle_graph: EagleGraph):
url = reverse("eagle-detail", kwargs={"pk": eagle_graph.eagle.pk})
response = api_client.get(url, {"select_related": "location", "mark_considered": "location"})
assert response.status_code == 200

def test_mark_considered_before_query_suppresses(self):
warns = self.request(select_related="location", mark_considered="location", mark_before="1")
assert warns == []
def test_mark_considered_before_query_suppresses(self, api_client: APIClient, eagle_graph: EagleGraph):
url = reverse("eagle-detail", kwargs={"pk": eagle_graph.eagle.pk})
response = api_client.get(
url, {"select_related": "location", "mark_considered": "location", "mark_before": "1"}
)
assert response.status_code == 200

def test_mark_considered_multiple_cache_names(self):
warns = self.request(
select_related="location",
prefetch_related="previous_locations",
mark_considered="location,previous_locations",
def test_mark_considered_multiple_cache_names(self, api_client: APIClient, eagle_graph: EagleGraph):
url = reverse("eagle-detail", kwargs={"pk": eagle_graph.eagle.pk})
response = api_client.get(
url,
{
"select_related": "location",
"prefetch_related": "previous_locations",
"mark_considered": "location,previous_locations",
},
)
assert warns == []
assert response.status_code == 200

def test_mark_considered_wrong_field_still_warns(self):
warns = self.request(select_related="location", mark_considered="previous_locations")
assert len(warns) == 1
assert 'select_related("location")' in str(warns[0])
def test_mark_considered_wrong_field_still_warns(self, api_client: APIClient, eagle_graph: EagleGraph):
url = reverse("eagle-detail", kwargs={"pk": eagle_graph.eagle.pk})
with pytest.raises(UnusedRelatedAccess) as exc_info:
api_client.get(url, {"select_related": "location", "mark_considered": "previous_locations"})
assert 'select_related("location")' in str(exc_info.value)
79 changes: 48 additions & 31 deletions tests/test_may_access.py
Original file line number Diff line number Diff line change
@@ -1,45 +1,59 @@
import asyncio

import pytest
from django.urls import reverse
from rest_framework.test import APIClient

import eagle
from eagle import mark_considered, may_access
from eagle import UnusedRelatedAccess, mark_considered, may_access, unused
from test_project.models import Eagle
from tests.base import (
BaseRequestTest,
EagleGraph,
EagleWithLocation,
InactiveCollectorTestBase,
MayAccessHelperTestBase,
)


class TestMayAccess(BaseRequestTest):
def test_may_access_suppresses_when_called(self):
warns = self.request(select_related="location", may_access="location", may_access_call="1")
assert warns == []

def test_may_access_does_not_suppress_when_not_called(self):
warns = self.request(select_related="location", may_access="location")
assert len(warns) == 1
assert 'select_related("location")' in str(warns[0])

def test_may_access_multiple_fields(self):
warns = self.request(
select_related="location",
prefetch_related="previous_locations",
may_access="location,previous_locations",
may_access_call="1",
def test_may_access_suppresses_when_called(self, api_client: APIClient, eagle_graph: EagleGraph):
url = reverse("eagle-detail", kwargs={"pk": eagle_graph.eagle.pk})
response = api_client.get(url, {"select_related": "location", "may_access": "location", "may_access_call": "1"})
assert response.status_code == 200

def test_may_access_does_not_suppress_when_not_called(self, api_client: APIClient, eagle_graph: EagleGraph):
url = reverse("eagle-detail", kwargs={"pk": eagle_graph.eagle.pk})
with pytest.raises(UnusedRelatedAccess) as exc_info:
api_client.get(url, {"select_related": "location", "may_access": "location"})
assert 'select_related("location")' in str(exc_info.value)

def test_may_access_multiple_fields(self, api_client: APIClient, eagle_graph: EagleGraph):
url = reverse("eagle-detail", kwargs={"pk": eagle_graph.eagle.pk})
response = api_client.get(
url,
{
"select_related": "location",
"prefetch_related": "previous_locations",
"may_access": "location,previous_locations",
"may_access_call": "1",
},
)
assert warns == []

def test_may_access_does_not_mark_when_wrapped_raises(self):
warns = self.request(
select_related="location",
may_access="location",
may_access_call="1",
may_access_raise="1",
)
assert len(warns) == 1
assert 'select_related("location")' in str(warns[0])
assert response.status_code == 200

def test_may_access_does_not_mark_when_wrapped_raises(self, api_client: APIClient, eagle_graph: EagleGraph):
url = reverse("eagle-detail", kwargs={"pk": eagle_graph.eagle.pk})
with pytest.raises(UnusedRelatedAccess) as exc_info:
api_client.get(
url,
{
"select_related": "location",
"may_access": "location",
"may_access_call": "1",
"may_access_raise": "1",
},
)
assert 'select_related("location")' in str(exc_info.value)


class TestPublicExport:
Expand All @@ -49,17 +63,18 @@ def test_public_exports(self):


class TestMayAccessHelper(MayAccessHelperTestBase):
def test_may_access_async_function_suppresses_after_await(self):
def test_may_access_async_function_suppresses_after_await(self, eagle_with_location: EagleWithLocation):
Eagle.objects.select_related("location").get()

@may_access(Eagle, "location")
async def consumer():
return 7

assert asyncio.run(consumer()) == 7
assert self.flush() == []
# location was marked accessed after the await, so ending the request emits no warning (no error).
unused.end_request()

def test_may_access_async_function_no_mark_when_raises(self):
def test_may_access_async_function_no_mark_when_raises(self, eagle_with_location: EagleWithLocation):
Eagle.objects.select_related("location").get()

class Boom(Exception):
Expand All @@ -71,7 +86,9 @@ async def consumer():

with pytest.raises(Boom):
asyncio.run(consumer())
assert len(self.flush()) == 1
# The wrapped function raised before marking, so the loaded relation is still unused.
with pytest.raises(UnusedRelatedAccess):
unused.end_request()


class TestMayAccessInactiveCollector(InactiveCollectorTestBase):
Expand Down
Loading
Loading