255 lines
10 KiB
Python
255 lines
10 KiB
Python
"""
|
|
Tests for tenant ownership validation in basic-info CRUD operations.
|
|
|
|
Per AGENTS.md: "All business queries SHALL filter by current user's tenant_no."
|
|
Per audit: customer, cell_sample_type, collection_protocol_no, sample_code
|
|
CRUD get_*() functions do NOT filter by tenant_no, allowing cross-tenant
|
|
update/delete if the record ID is known.
|
|
|
|
These tests verify that:
|
|
1. get_*() functions accept and filter by tenant_no
|
|
2. update_*() functions reject cross-tenant modifications
|
|
3. soft_delete_*() functions reject cross-tenant deletions
|
|
|
|
The collection_region module already correctly implements tenant_no filtering
|
|
and serves as the reference pattern.
|
|
"""
|
|
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
# We test the CRUD function signatures and behavior without a real database
|
|
# by patching the SQLAlchemy session and model queries.
|
|
|
|
|
|
class TestCustomerTenantOwnership(unittest.TestCase):
|
|
"""Customer CRUD should enforce tenant_no on update/delete."""
|
|
|
|
def test_get_customer_signature_accepts_tenant_no(self):
|
|
"""get_customer() should accept an optional tenant_no parameter."""
|
|
from app.crud.customer import get_customer
|
|
|
|
import inspect
|
|
sig = inspect.signature(get_customer)
|
|
params = list(sig.parameters.keys())
|
|
# Currently: ['db', 'customer_id'] — should add 'tenant_no'
|
|
self.assertIn("tenant_no", params, "get_customer() must accept tenant_no parameter")
|
|
|
|
def test_get_customer_filters_by_tenant_no(self):
|
|
"""get_customer() should filter query by tenant_no when provided."""
|
|
from app.crud.customer import get_customer
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_db.query.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.first.return_value = None
|
|
|
|
# Should not raise, and should add tenant_no filter
|
|
result = get_customer(mock_db, "some-id", tenant_no="tenant-A")
|
|
mock_db.query.assert_called_once()
|
|
# filter() should be called at least twice (del_flag + tenant_no)
|
|
self.assertGreaterEqual(mock_query.filter.call_count, 2)
|
|
|
|
def test_update_customer_rejects_cross_tenant(self):
|
|
"""update_customer() should raise when record belongs to different tenant."""
|
|
from app.crud.customer import update_customer
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_db.query.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.first.return_value = None # tenant_no filter blocks access
|
|
|
|
with self.assertRaises(ValueError) as ctx:
|
|
update_customer(
|
|
mock_db,
|
|
"cross-tenant-id",
|
|
customer_name="Hacked",
|
|
user_id="user-A",
|
|
tenant_no="tenant-A", # record belongs to tenant-B
|
|
)
|
|
self.assertIn("不存在", str(ctx.exception))
|
|
|
|
def test_soft_delete_customer_rejects_cross_tenant(self):
|
|
"""soft_delete_customer() should raise when record belongs to different tenant."""
|
|
from app.crud.customer import soft_delete_customer
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_db.query.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.first.return_value = None # tenant_no filter blocks access
|
|
|
|
with self.assertRaises(ValueError):
|
|
soft_delete_customer(mock_db, "cross-tenant-id", tenant_no="tenant-A")
|
|
|
|
|
|
class TestCellSampleTypeTenantOwnership(unittest.TestCase):
|
|
"""CellSampleType CRUD should enforce tenant_no on update/delete."""
|
|
|
|
def test_get_cell_sample_type_signature_accepts_tenant_no(self):
|
|
"""get_cell_sample_type() should accept an optional tenant_no parameter."""
|
|
from app.crud.cell_sample_type import get_cell_sample_type
|
|
|
|
import inspect
|
|
sig = inspect.signature(get_cell_sample_type)
|
|
params = list(sig.parameters.keys())
|
|
self.assertIn("tenant_no", params, "get_cell_sample_type() must accept tenant_no parameter")
|
|
|
|
def test_update_cell_sample_type_rejects_cross_tenant(self):
|
|
"""update_cell_sample_type() should raise when record belongs to different tenant."""
|
|
from app.crud.cell_sample_type import update_cell_sample_type
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_db.query.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.first.return_value = None
|
|
|
|
with self.assertRaises(ValueError) as ctx:
|
|
update_cell_sample_type(
|
|
mock_db,
|
|
"cross-tenant-id",
|
|
type_name="Hacked",
|
|
user_id="user-A",
|
|
tenant_no="tenant-A",
|
|
)
|
|
self.assertIn("不存在", str(ctx.exception))
|
|
|
|
def test_soft_delete_cell_sample_type_rejects_cross_tenant(self):
|
|
"""soft_delete_cell_sample_type() should raise when record belongs to different tenant."""
|
|
from app.crud.cell_sample_type import soft_delete_cell_sample_type
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_db.query.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.first.return_value = None
|
|
|
|
with self.assertRaises(ValueError):
|
|
soft_delete_cell_sample_type(mock_db, "cross-tenant-id", tenant_no="tenant-A")
|
|
|
|
|
|
class TestProtocolNoTenantOwnership(unittest.TestCase):
|
|
"""CollectionProtocolNo CRUD should enforce tenant_no on update/delete."""
|
|
|
|
def test_get_protocol_no_signature_accepts_tenant_no(self):
|
|
"""get_protocol_no() should accept an optional tenant_no parameter."""
|
|
from app.crud.collection_protocol_no import get_protocol_no
|
|
|
|
import inspect
|
|
sig = inspect.signature(get_protocol_no)
|
|
params = list(sig.parameters.keys())
|
|
self.assertIn("tenant_no", params, "get_protocol_no() must accept tenant_no parameter")
|
|
|
|
def test_update_protocol_no_rejects_cross_tenant(self):
|
|
"""update_protocol_no() should raise when record belongs to different tenant."""
|
|
from app.crud.collection_protocol_no import update_protocol_no
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_db.query.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.first.return_value = None
|
|
|
|
with self.assertRaises(ValueError) as ctx:
|
|
update_protocol_no(
|
|
mock_db,
|
|
"cross-tenant-id",
|
|
remark="Hacked",
|
|
user_id="user-A",
|
|
tenant_no="tenant-A",
|
|
)
|
|
self.assertIn("不存在", str(ctx.exception))
|
|
|
|
def test_soft_delete_protocol_no_rejects_cross_tenant(self):
|
|
"""soft_delete_protocol_no() should raise when record belongs to different tenant."""
|
|
from app.crud.collection_protocol_no import soft_delete_protocol_no
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_db.query.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.first.return_value = None
|
|
|
|
with self.assertRaises(ValueError):
|
|
soft_delete_protocol_no(mock_db, "cross-tenant-id", tenant_no="tenant-A")
|
|
|
|
|
|
class TestSampleCodeTenantOwnership(unittest.TestCase):
|
|
"""SampleCode CRUD should enforce tenant_no on update/delete."""
|
|
|
|
def test_get_sample_code_signature_accepts_tenant_no(self):
|
|
"""get_sample_code() should accept an optional tenant_no parameter."""
|
|
from app.crud.sample_code import get_sample_code
|
|
|
|
import inspect
|
|
sig = inspect.signature(get_sample_code)
|
|
params = list(sig.parameters.keys())
|
|
self.assertIn("tenant_no", params, "get_sample_code() must accept tenant_no parameter")
|
|
|
|
def test_update_sample_code_rejects_cross_tenant(self):
|
|
"""update_sample_code() should raise when record belongs to different tenant."""
|
|
from app.crud.sample_code import update_sample_code
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_db.query.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.first.return_value = None
|
|
|
|
with self.assertRaises(ValueError) as ctx:
|
|
update_sample_code(
|
|
mock_db,
|
|
"cross-tenant-id",
|
|
remark="Hacked",
|
|
user_id="user-A",
|
|
tenant_no="tenant-A",
|
|
)
|
|
self.assertIn("不存在", str(ctx.exception))
|
|
|
|
def test_soft_delete_sample_code_rejects_cross_tenant(self):
|
|
"""soft_delete_sample_code() should raise when record belongs to different tenant."""
|
|
from app.crud.sample_code import soft_delete_sample_code
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_db.query.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.first.return_value = None
|
|
|
|
with self.assertRaises(ValueError):
|
|
soft_delete_sample_code(mock_db, "cross-tenant-id", tenant_no="tenant-A")
|
|
|
|
|
|
class TestGetTenantNoFallback(unittest.TestCase):
|
|
"""_get_tenant_no() should not silently fall back to hardcoded value."""
|
|
|
|
def test_get_tenant_no_raises_on_missing_user(self):
|
|
"""When user_id is missing from token, should raise rather than default."""
|
|
from app.api.basic_info import _get_tenant_no
|
|
|
|
mock_db = MagicMock()
|
|
user = {} # no user_id
|
|
with self.assertRaises((ValueError, PermissionError)):
|
|
_get_tenant_no(mock_db, user)
|
|
|
|
def test_get_tenant_no_raises_on_missing_sys_user(self):
|
|
"""When SysUser record not found in DB, should raise rather than default."""
|
|
from app.api.basic_info import _get_tenant_no
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_db.query.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.first.return_value = None # SysUser not found
|
|
|
|
user = {"user_id": "some-id"}
|
|
with self.assertRaises((ValueError, PermissionError)):
|
|
_get_tenant_no(mock_db, user)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|