134 lines
4.5 KiB
Python
134 lines
4.5 KiB
Python
"""
|
|
tests/test_tenant_isolation.py
|
|
------------------------------
|
|
Cross-tenant data-isolation guard (MT-1 core mechanism).
|
|
|
|
The entire database-per-tenant isolation guarantee rests on ONE thing:
|
|
`RoutingSession.get_bind()` binding every query to `g.tenant_engine` for the
|
|
current request. If a future refactor breaks that — returns a stale engine, or
|
|
falls back to the default when a tenant is active — tenant A would silently read
|
|
or write tenant B's database. That is the single worst, hardest-to-detect
|
|
failure mode for the SaaS, so it gets a dedicated test.
|
|
|
|
These tests construct two independent in-memory SQLite databases (standing in
|
|
for two tenant DBs), seed each with distinct data, then flip `g.tenant_engine`
|
|
and assert `db.session` reads and writes land in — and only in — the selected
|
|
tenant's database. No MySQL required; the routing logic is engine-agnostic.
|
|
"""
|
|
|
|
import pytest
|
|
from flask import g
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.pool import StaticPool
|
|
from sqlalchemy.orm import Session as SASession
|
|
|
|
|
|
def _make_tenant_engine():
|
|
"""A standalone in-memory SQLite engine that persists across connections.
|
|
|
|
StaticPool keeps a single underlying connection so the in-memory schema and
|
|
rows survive between checkouts (a plain sqlite:// memory DB is per-connection
|
|
and would appear empty on the next query).
|
|
"""
|
|
return create_engine(
|
|
'sqlite://',
|
|
connect_args={'check_same_thread': False},
|
|
poolclass=StaticPool,
|
|
future=True,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def two_tenants(app):
|
|
"""Two isolated tenant DBs: A has facility 'ACME-HQ', B has 'Globex-Plant'."""
|
|
from app import db
|
|
from app.models.facility import Facility
|
|
|
|
eng_a = _make_tenant_engine()
|
|
eng_b = _make_tenant_engine()
|
|
|
|
with app.app_context():
|
|
db.metadata.create_all(eng_a)
|
|
db.metadata.create_all(eng_b)
|
|
|
|
sa = SASession(eng_a)
|
|
sa.add(Facility(name='ACME-HQ', active=True))
|
|
sa.commit(); sa.close()
|
|
|
|
sb = SASession(eng_b)
|
|
sb.add(Facility(name='Globex-Plant', active=True))
|
|
sb.commit(); sb.close()
|
|
|
|
yield eng_a, eng_b
|
|
|
|
eng_a.dispose()
|
|
eng_b.dispose()
|
|
|
|
|
|
def _facility_names():
|
|
from app import db
|
|
from app.models.facility import Facility
|
|
return {f.name for f in db.session.query(Facility).all()}
|
|
|
|
|
|
def test_reads_are_routed_to_the_active_tenant(app, two_tenants):
|
|
"""db.session reads only ever see the tenant bound via g.tenant_engine."""
|
|
from app import db
|
|
eng_a, eng_b = two_tenants
|
|
|
|
with app.app_context():
|
|
# Bind tenant A.
|
|
g.tenant_engine = eng_a
|
|
db.session.remove() # force a fresh bind on next query
|
|
names = _facility_names()
|
|
assert names == {'ACME-HQ'}
|
|
assert 'Globex-Plant' not in names # B's data must never leak into A
|
|
|
|
# Switch to tenant B — same session machinery, different engine.
|
|
g.tenant_engine = eng_b
|
|
db.session.remove()
|
|
names = _facility_names()
|
|
assert names == {'Globex-Plant'}
|
|
assert 'ACME-HQ' not in names # A's data must never leak into B
|
|
|
|
|
|
def test_writes_do_not_leak_across_tenants(app, two_tenants):
|
|
"""A write performed while tenant B is active must not touch tenant A."""
|
|
from app import db
|
|
from app.models.facility import Facility
|
|
eng_a, eng_b = two_tenants
|
|
|
|
with app.app_context():
|
|
# Insert into B.
|
|
g.tenant_engine = eng_b
|
|
db.session.remove()
|
|
db.session.add(Facility(name='Globex-NewSite', active=True))
|
|
db.session.commit()
|
|
|
|
# A must be completely unaffected by B's write.
|
|
g.tenant_engine = eng_a
|
|
db.session.remove()
|
|
names_a = _facility_names()
|
|
assert names_a == {'ACME-HQ'}
|
|
assert 'Globex-NewSite' not in names_a
|
|
|
|
# And B genuinely received the new row.
|
|
g.tenant_engine = eng_b
|
|
db.session.remove()
|
|
names_b = _facility_names()
|
|
assert 'Globex-NewSite' in names_b
|
|
|
|
|
|
def test_routing_reads_g_dynamically_per_request(app, two_tenants):
|
|
"""Re-binding g.tenant_engine within the app context re-routes subsequent
|
|
queries — proving get_bind() reads g live, not a cached value."""
|
|
from app import db
|
|
eng_a, eng_b = two_tenants
|
|
|
|
with app.app_context():
|
|
for engine, expected in ((eng_a, 'ACME-HQ'), (eng_b, 'Globex-Plant'),
|
|
(eng_a, 'ACME-HQ')):
|
|
g.tenant_engine = engine
|
|
db.session.remove()
|
|
assert _facility_names() == {expected}
|