Testing¶
django-rls-tenants provides test helpers for setting up RLS context and verifying that RLS policies are correctly applied.
Context Managers¶
rls_bypass¶
Enables admin bypass for the duration of the block. Useful for test setup and teardown where you need to create/read data across tenants:
from django_rls_tenants.tenants.testing import rls_bypass
def test_order_creation():
with rls_bypass():
# Create test data visible to all tenants
tenant = Tenant.objects.create(name="Acme")
Order.objects.create(tenant=tenant, title="Test", amount=100)
# Verify data exists
assert Order.objects.count() == 1
Parameters:
| Parameter | Type | Default | Description |
|---|---|---|---|
using |
str |
"default" |
Database alias. |
rls_as_tenant¶
Scopes queries to a specific tenant. Useful for testing tenant isolation:
from django_rls_tenants.tenants.testing import rls_as_tenant, rls_bypass
def test_tenant_isolation():
with rls_bypass():
t1 = Tenant.objects.create(name="Tenant 1")
t2 = Tenant.objects.create(name="Tenant 2")
Order.objects.create(tenant=t1, title="T1 Order", amount=100)
Order.objects.create(tenant=t2, title="T2 Order", amount=200)
with rls_as_tenant(tenant_id=t1.pk):
orders = list(Order.objects.all())
assert len(orders) == 1
assert orders[0].title == "T1 Order"
with rls_as_tenant(tenant_id=t2.pk):
orders = list(Order.objects.all())
assert len(orders) == 1
assert orders[0].title == "T2 Order"
Parameters:
| Parameter | Type | Default | Description |
|---|---|---|---|
tenant_id |
int \| str |
(required) | Tenant PK to scope to. |
using |
str |
"default" |
Database alias. |
Auto-scoping is active
Inside an rls_as_tenant block, RLSManager.get_queryset() automatically adds
WHERE tenant_id = X to all queries. This means Order.objects.all() returns
only the scoped tenant's rows at both ORM and RLS levels.
Assertion Functions¶
assert_rls_enabled¶
Verifies that RLS is enabled and forced on a table:
from django_rls_tenants.tenants.testing import assert_rls_enabled
def test_rls_enabled_on_orders():
assert_rls_enabled("myapp_order")
Raises AssertionError if:
- The table does not exist.
- RLS is not enabled (
relrowsecurityisFalse). - RLS is not forced (
relforcerowsecurityisFalse).
assert_rls_policy_exists¶
Verifies that an RLS policy exists on a table:
from django_rls_tenants.tenants.testing import assert_rls_policy_exists
def test_isolation_policy_exists():
assert_rls_policy_exists("myapp_order")
# Custom policy name:
assert_rls_policy_exists("myapp_order", policy_name="myapp_order_tenant_isolation_policy")
Default policy name: "{table_name}_tenant_isolation_policy"
assert_rls_blocks_without_context¶
Verifies the fail-closed behavior -- queries without RLS context return zero rows:
from django_rls_tenants.tenants.testing import assert_rls_blocks_without_context, rls_bypass
def test_fail_closed():
with rls_bypass():
# Must have at least one row for the assertion to be meaningful
Order.objects.create(tenant=tenant, title="Test", amount=100)
# Without any RLS context, zero rows should be returned
assert_rls_blocks_without_context(Order)
This function:
- Enters
admin_context()to verify the table has at least one row (prevents vacuous passes on empty tables). - Clears the admin and tenant GUCs on the
usingconnection, counts the rows again and restores the GUCs to their previous values (withSET LOCALwhenUSE_LOCAL_SET=Trueinsidetransaction.atomic()). The restore happens before the assertion, so the caller's context is intact even when the assertion fails. WithUSE_LOCAL_SET=Trueoutsideatomic()(1.4.0+), the count runs with the GUC affinity guard seeing no context, and the GUCs are cleared and restored on the session. - Asserts that the second count is zero.
Both counts are raw SQL on the model's table (SELECT COUNT(*)), not ORM queries. The
result therefore depends only on the database: the STRICT_MODE guard, the auto-scope
filter and the filters of the model's default manager are not involved.
Changed in 1.4.0. The helper works with STRICT_MODE=True, and its result no longer
depends on the admin or tenant context it is called from. Inside rls_bypass() or
rls_as_tenant() it checks the same no-context state and leaves that context intact.
Before, it raised NoTenantContextError under strict mode, and inside rls_bypass() or
rls_as_tenant() it counted with the caller's context and failed with "RLS may not be
enforced". Only the admin and tenant GUCs are cleared: a flag set with bypass_flag()
stays set and can make rows visible.
Raises AssertionError if:
- Rows are returned (RLS is not enforcing isolation). Common causes: the test connects as
a superuser or a role with
BYPASSRLS(see W005), or RLS is not enabled and forced on the table. - The table is empty (would pass vacuously).
Raises RuntimeError if:
USE_LOCAL_SET=True,usingis not inRLS_TENANTS["DATABASES"]and its connection is not insidetransaction.atomic(). Before 1.4.0 it raised for every alias outsideatomic().
pytest Integration¶
Using Fixtures¶
Create reusable fixtures for test data setup:
import pytest
from django_rls_tenants.tenants.testing import rls_bypass, rls_as_tenant
@pytest.fixture
def tenant_a():
with rls_bypass():
return Tenant.objects.create(name="Tenant A")
@pytest.fixture
def tenant_b():
with rls_bypass():
return Tenant.objects.create(name="Tenant B")
@pytest.fixture
def sample_orders(tenant_a, tenant_b):
with rls_bypass():
Order.objects.create(tenant=tenant_a, title="A1", amount=100)
Order.objects.create(tenant=tenant_a, title="A2", amount=200)
Order.objects.create(tenant=tenant_b, title="B1", amount=300)
Marking Tests¶
Use pytest.mark.django_db for tests that need database access:
import pytest
pytestmark = pytest.mark.django_db
def test_tenant_sees_own_orders(sample_orders, tenant_a):
with rls_as_tenant(tenant_id=tenant_a.pk):
assert Order.objects.count() == 2
def test_no_context_sees_nothing(sample_orders):
assert_rls_blocks_without_context(Order)
For tests that require transaction isolation:
@pytest.mark.django_db(transaction=True)
def test_rls_with_transactions(tenant_a):
# Tests that need real transaction boundaries
...
Async Tests¶
Added in 1.4.0. Every helper has an async counterpart for async def tests, in the same
module and with the same arguments:
| Sync helper | Async helper | Use it with |
|---|---|---|
rls_bypass() |
async_rls_bypass() |
async with |
rls_as_tenant() |
async_rls_as_tenant() |
async with |
assert_rls_enabled() |
aassert_rls_enabled() |
await |
assert_rls_policy_exists() |
aassert_rls_policy_exists() |
await |
assert_rls_blocks_without_context() |
aassert_rls_blocks_without_context() |
await |
The two context managers wrap
async_admin_context() and async_tenant_context(). They
set the tenant state in the test coroutine, so the auto-scope filter applies to the async
ORM (acount(), aget(), async for). Each assertion runs its sync version in one
thread_sensitive sync_to_async hop.
Which connection an async test uses¶
Django has no asynchronous database cursor. The helpers and the async ORM run their SQL in
thread_sensitive hops. The thread a hop lands on determines its connection, and therefore
the transaction it sees. That thread depends on how the test runs the coroutine:
async defmethods on a DjangoTestCase(recommended). Django runs the method throughasync_to_sync, so the hops run on the test thread. They use the test's connection and transaction, they see rows fromsetUpTestDataand fixtures, andUSE_LOCAL_SET=Trueworks.async defmethods on aTransactionTestCase. The hops also run on the test thread and use its connection, but no test transaction is open. Writes are committed, and the tables are flushed after each test. WithUSE_LOCAL_SET=True, the async context managers write nothing there and the guard gives each statement a transaction of its own.- A sync pytest test that drives a coroutine with
async_to_sync. The hops run on the test thread too. With plain@pytest.mark.django_dbthis behaves like aTestCaseand works with your usual fixtures. Withtransaction=Trueit behaves like aTransactionTestCase. The library tests itself this way. async deftests run by pytest-asyncio, or by anyio's pytest plugin on the asyncio backend. These plugins run the event loop directly on the test thread, with noasync_to_syncabove it. Every hop therefore goes to asgiref's shared executor thread, which opens a separate connection. See the warning below.
from django.test import TestCase
from django_rls_tenants.tenants.testing import (
aassert_rls_blocks_without_context,
async_rls_as_tenant,
async_rls_bypass,
rls_bypass,
)
class OrderIsolationTests(TestCase):
@classmethod
def setUpTestData(cls):
with rls_bypass():
cls.tenant_a = Tenant.objects.create(name="Tenant A")
tenant_b = Tenant.objects.create(name="Tenant B")
Order.objects.create(tenant=cls.tenant_a, title="A1", amount=100)
Order.objects.create(tenant=tenant_b, title="B1", amount=200)
async def test_tenant_sees_own_orders(self):
async with async_rls_as_tenant(tenant_id=self.tenant_a.pk):
titles = [order.title async for order in Order.objects.all()]
assert titles == ["A1"]
async def test_admin_sees_all(self):
async with async_rls_bypass():
assert await Order.objects.acount() == 2
async def test_fail_closed(self):
await aassert_rls_blocks_without_context(Order)
Like its sync version, aassert_rls_blocks_without_context() passes only when the test
connects with a role that RLS applies to. A superuser, or a role with BYPASSRLS, sees
every row.
import pytest
from asgiref.sync import async_to_sync
from django_rls_tenants.tenants.testing import async_rls_as_tenant
pytestmark = pytest.mark.django_db
def test_async_report_is_scoped(tenant_a, sample_orders):
async def run():
async with async_rls_as_tenant(tenant_id=tenant_a.pk):
return await build_order_report() # the async code under test
assert async_to_sync(run)()["count"] == 2
pytest-asyncio and anyio tests use a separate connection
In an async def test run by pytest-asyncio or anyio's pytest plugin, the helpers and
the async ORM use asgiref's executor thread and its own database connection, outside the
test's transaction. As a result:
- With plain
django_db, rows that fixtures create inside the test transaction are invisible to the test. - Rows the test writes are committed, so they outlive the test and later tests see them.
- A write that collides with a row the test transaction inserted or locked (for
example
aget_or_create()on a fixture row's unique key) waits for that transaction to end. It cannot end while the test runs, so the test hangs. - Anything a sync fixture set on the test's connection, such as
SET ROLEor GUCs, does not apply. If RLS is only enforced after such aSET ROLE(because the tests connect as a superuser), it is bypassed on the executor connection. ORM-level isolation checks still pass there, because of the auto-scope filter, so they pass for the wrong reason. - The extra connection stays open, so pytest-django cannot drop the test database at
the end of the session. It reports this only as a
PytestWarning("database ... is being accessed by other users") and leaves the database behind.
Prefer the patterns above. If you must use the plugin, mark the tests
@pytest.mark.django_db(transaction=True). Committed fixture rows are then visible,
nothing waits on an open test transaction, and the tables are flushed after each test.
The SET ROLE point still applies. Also close the executor thread's
connection when the test ends:
import pytest
from asgiref.sync import sync_to_async
from django.db import connections
from django_rls_tenants.tenants.testing import async_rls_as_tenant
@pytest.mark.asyncio
@pytest.mark.django_db(transaction=True)
async def test_with_pytest_asyncio(tenant_a):
try:
async with async_rls_as_tenant(tenant_id=tenant_a.pk):
...
finally:
await sync_to_async(connections.close_all, thread_sensitive=True)()
Strict Mode in Tests¶
When STRICT_MODE=True, any query on an RLS-protected model without an active
context raises NoTenantContextError. This affects test setup code that queries
models outside a context.
Use rls_bypass() (which wraps admin_context()) or rls_as_tenant() in
fixture and setup code:
@pytest.fixture
def sample_data():
# rls_bypass establishes an active context -- passes strict mode check
with rls_bypass():
tenant = Tenant.objects.create(name="Test")
Order.objects.create(tenant=tenant, title="Order 1", amount=100)
return tenant
The assertion functions work under strict mode. assert_rls_enabled() and
assert_rls_policy_exists() only read the PostgreSQL catalog, and
assert_rls_blocks_without_context() counts rows with raw SQL, which the STRICT_MODE
check does not cover. You do not have to turn STRICT_MODE off to use these helpers.
To explicitly test that strict mode raises, use pytest.raises:
from django_rls_tenants.exceptions import NoTenantContextError
@override_settings(RLS_TENANTS={**RLS_SETTINGS, "STRICT_MODE": True})
def test_strict_mode_raises_without_context():
with pytest.raises(NoTenantContextError, match="strict mode"):
Order.objects.count()
Overriding RLS_TENANTS Per Test¶
Any of Django's override mechanisms works for RLS_TENANTS: the override_settings
context manager or decorator, SimpleTestCase.settings(), and pytest-django's
settings fixture. The library listens for Django's setting_changed signal and
re-reads the dict the next time a setting is accessed, so a test can flip runtime
options such as STRICT_MODE or USE_LOCAL_SET without manually clearing the config
cache. Assign a new dict, preserving any other required settings:
def test_local_set_mode(settings, tenant_a):
settings.RLS_TENANTS = {**settings.RLS_TENANTS, "USE_LOCAL_SET": True}
with transaction.atomic(), rls_as_tenant(tenant_id=tenant_a.pk):
...
Since 1.4.0 the transaction.atomic() is optional on an alias in
RLS_TENANTS["DATABASES"]: outside a transaction, the guard gives each statement the
context in a transaction of its own (see
Code Outside a Request Transaction).
Reloading the config does not rebuild model fields or database RLS policies.
Overrides of schema-related options such as TENANT_MODEL, TENANT_FK_FIELD,
TENANT_PK_TYPE or GUC_PREFIX require matching models and policies. In particular,
changing GUC_PREFIX alone makes the library set different variables from those
read by the existing policies; see the
GUC_PREFIX configuration guidance.
Before 1.3.2
Versions up to 1.3.1 cached RLS_TENANTS on first read and never invalidated it,
so these overrides were silently ignored and the test ran with the original
values. If a test suite worked around that by clearing
rls_tenants_config._config_cache, the workaround can be removed.
Multi-Database Testing¶
All test helpers accept a using parameter:
with rls_bypass(using="replica"):
data = Order.objects.using("replica").all()
assert_rls_enabled("myapp_order", using="replica")
Changed in 1.4.0: rls_bypass() and rls_as_tenant() apply to every alias in
RLS_TENANTS["DATABASES"], whichever one using names, in both GUC modes (see
Multi-Database Support). An alias outside the
list gets the context only when using names it, and the configured aliases follow it too.
With USE_LOCAL_SET=True, naming such an alias raises RuntimeError
(RLSConfigurationError for the async helpers) unless the helper runs inside
transaction.atomic(using=...).
Example: Full Test Suite¶
import pytest
from django_rls_tenants.tenants.testing import (
assert_rls_blocks_without_context,
assert_rls_enabled,
assert_rls_policy_exists,
rls_as_tenant,
rls_bypass,
)
pytestmark = pytest.mark.django_db
class TestOrderRLS:
def test_rls_enabled(self):
assert_rls_enabled("myapp_order")
def test_policy_exists(self):
assert_rls_policy_exists("myapp_order")
def test_fail_closed(self, sample_orders):
assert_rls_blocks_without_context(Order)
def test_tenant_isolation(self, tenant_a, tenant_b, sample_orders):
with rls_as_tenant(tenant_id=tenant_a.pk):
a_orders = list(Order.objects.values_list("title", flat=True))
assert sorted(a_orders) == ["A1", "A2"]
with rls_as_tenant(tenant_id=tenant_b.pk):
b_orders = list(Order.objects.values_list("title", flat=True))
assert b_orders == ["B1"]
def test_admin_sees_all(self, sample_orders):
with rls_bypass():
assert Order.objects.count() == 3