zkxw/backend/tests/test_basic_info_tenant_ownership.py

"""
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()