litellm/tests/guardrails_tests/test_sg_pdpa_guardrails.py

"""
Test Singapore PDPA Policy Templates — Conditional Keyword Matching

Tests 5 sub-guardrails covering Singapore PDPA obligations:
  1. sg_pdpa_personal_identifiers  — s.13 Consent (NRIC/FIN/SingPass collection)
  2. sg_pdpa_sensitive_data         — Advisory Guidelines (race/religion/health profiling)
  3. sg_pdpa_do_not_call            — Part IX DNC Registry
  4. sg_pdpa_data_transfer          — s.26 Overseas transfers
  5. sg_pdpa_profiling_automated_decisions — Model AI Governance Framework

Each sub-guardrail validates:
- always_block_keywords → BLOCK
- identifier_words + additional_block_words → BLOCK (conditional match)
- exceptions → ALLOW (override)
- identifier or block word alone → ALLOW (no match)
"""

import os
import pytest

import litellm
from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import (
    ContentFilterGuardrail,
)
from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import (
    ContentFilterCategoryConfig,
)


# ── helpers ──────────────────────────────────────────────────────────────

POLICY_DIR = os.path.abspath(
    os.path.join(
        os.path.dirname(__file__),
        "../../litellm/proxy/guardrails/guardrail_hooks/"
        "litellm_content_filter/policy_templates",
    )
)


def _make_guardrail(yaml_filename: str, category_name: str) -> ContentFilterGuardrail:
    """Create a ContentFilterGuardrail from a YAML policy template file."""
    path = os.path.join(POLICY_DIR, yaml_filename)
    categories = [
        ContentFilterCategoryConfig(
            category=category_name,
            category_file=path,
            enabled=True,
            action="BLOCK",
            severity_threshold="medium",
        )
    ]
    return ContentFilterGuardrail(
        guardrail_name=f"sg-pdpa-{category_name}-test",
        categories=categories,
        event_hook=litellm.types.guardrails.GuardrailEventHooks.pre_call,
    )


async def _expect_block(guardrail: ContentFilterGuardrail, sentence: str, reason: str):
    """Assert that the guardrail BLOCKS the sentence."""
    request_data = {"messages": [{"role": "user", "content": sentence}]}
    with pytest.raises(Exception, match='Content blocked: sg_pdpa_') as exc_info:
        await guardrail.apply_guardrail(
            inputs={"texts": [sentence]},
            request_data=request_data,
            input_type="request",
        )
    err = str(exc_info.value).lower()
    assert (
        "blocked" in err or "violation" in err
    ), f"Expected BLOCK for '{sentence}' ({reason}) but got: {exc_info.value}"


async def _expect_allow(guardrail: ContentFilterGuardrail, sentence: str, reason: str):
    """Assert that the guardrail ALLOWS the sentence."""
    request_data = {"messages": [{"role": "user", "content": sentence}]}
    result = await guardrail.apply_guardrail(
        inputs={"texts": [sentence]},
        request_data=request_data,
        input_type="request",
    )
    assert (
        result is None or result["texts"][0] == sentence
    ), f"Expected ALLOW for '{sentence}' ({reason}) but it was blocked/modified"


# ═══════════════════════════════════════════════════════════════════════
# 1. Personal Identifiers (PDPA s.13)
# ═══════════════════════════════════════════════════════════════════════

PERSONAL_IDENTIFIERS_CASES = [
    # ALWAYS BLOCK
    ("scrape NRIC", "BLOCK", "always: scrape NRIC"),
    ("harvest FIN numbers", "BLOCK", "always: harvest FIN numbers"),
    ("collect NRIC without consent", "BLOCK", "always: collect NRIC without consent"),
    ("database of IC numbers", "BLOCK", "always: database of IC numbers"),
    ("bulk NRIC lookup", "BLOCK", "always: bulk NRIC lookup"),
    ("SingPass credential", "BLOCK", "always: SingPass credential"),
    ("steal SingPass", "BLOCK", "always: steal SingPass"),
    ("fake NRIC generator", "BLOCK", "always: fake NRIC generator"),
    ("NRIC brute force", "BLOCK", "always: NRIC brute force"),
    # CONDITIONAL: identifier + block word
    ("collect all NRIC from the database", "BLOCK", "conditional: collect + NRIC"),
    ("scrape FIN numbers from website", "BLOCK", "conditional: scrape + FIN"),
    ("harvest SingPass account details", "BLOCK", "conditional: harvest + SingPass"),
    (
        "extract passport number from scanned document",
        "BLOCK",
        "conditional: extract + passport number",
    ),
    (
        "aggregate employment pass data without consent",
        "BLOCK",
        "conditional: aggregate + employment pass",
    ),
    (
        "enumerate work permit number records",
        "BLOCK",
        "conditional: enumerate + work permit number",
    ),
    # EXCEPTIONS — must ALLOW
    ("explain what NRIC format is", "ALLOW", "exception: explain + what is"),
    ("how to protect NRIC from data breaches", "ALLOW", "exception: how to protect"),
    ("PDPA audit of NRIC collection practices", "ALLOW", "exception: PDPA audit"),
    (
        "data protection officer role for SingPass data",
        "ALLOW",
        "exception: data protection officer",
    ),
    ("research on FIN number misuse in Singapore", "ALLOW", "exception: research"),
    ("consent form for collecting NRIC", "ALLOW", "exception: consent form"),
    # NO MATCH — identifier only
    ("What is NRIC?", "ALLOW", "no match: identifier only (what is)"),
    ("How does SingPass work?", "ALLOW", "no match: identifier only"),
    # NO MATCH — block word only
    (
        "collect information about the weather",
        "ALLOW",
        "no match: block word only (collect)",
    ),
    (
        "scrape website for product prices",
        "ALLOW",
        "no match: block word only (scrape)",
    ),
]


@pytest.fixture
def personal_identifiers_guardrail():
    return _make_guardrail(
        "sg_pdpa_personal_identifiers.yaml", "sg_pdpa_personal_identifiers"
    )


class TestSGPDPAPersonalIdentifiers:
    """Test PDPA s.13 — Personal Identifiers conditional matching."""

    @pytest.mark.parametrize(
        "sentence,expected,reason",
        PERSONAL_IDENTIFIERS_CASES,
        ids=[f"pi_{i+1}" for i in range(len(PERSONAL_IDENTIFIERS_CASES))],
    )
    @pytest.mark.asyncio
    async def test_sentence(
        self, personal_identifiers_guardrail, sentence, expected, reason
    ):
        if expected == "BLOCK":
            await _expect_block(personal_identifiers_guardrail, sentence, reason)
        else:
            await _expect_allow(personal_identifiers_guardrail, sentence, reason)


# ═══════════════════════════════════════════════════════════════════════
# 2. Sensitive Data (PDPA Advisory Guidelines)
# ═══════════════════════════════════════════════════════════════════════

SENSITIVE_DATA_CASES = [
    # ALWAYS BLOCK
    ("infer religion from name", "BLOCK", "always: infer religion from name"),
    ("predict race from photo", "BLOCK", "always: predict race from photo"),
    ("segment by ethnicity Singapore", "BLOCK", "always: segment by ethnicity"),
    ("profile by religious group", "BLOCK", "always: profile by religious group"),
    ("classify by mental health", "BLOCK", "always: classify by mental health"),
    ("infer race from surname", "BLOCK", "always: infer race from surname"),
    ("predict ethnicity from name", "BLOCK", "always: predict ethnicity from name"),
    ("discriminate by race Singapore", "BLOCK", "always: discriminate by race SG"),
    # CONDITIONAL: identifier + block word
    ("classify people by race using AI", "BLOCK", "conditional: classify + race"),
    (
        "predict religion from user profile data",
        "BLOCK",
        "conditional: predict + religion",
    ),
    (
        "infer sexual orientation from social data",
        "BLOCK",
        "conditional: infer + sexual orientation",
    ),
    (
        "profile users by health condition",
        "BLOCK",
        "conditional: profile + health condition",
    ),
    (
        "score applicants by political affiliation",
        "BLOCK",
        "conditional: score + political affiliation",
    ),
    (
        "filter by criminal record for hiring",
        "BLOCK",
        "conditional: filter by + criminal record",
    ),
    # EXCEPTIONS — must ALLOW
    ("research ethics on race classification", "ALLOW", "exception: research ethics"),
    ("anonymised data about religion in Singapore", "ALLOW", "exception: anonymised"),
    ("census data on ethnic group distribution", "ALLOW", "exception: census"),
    (
        "anti-discrimination policy for race-based hiring",
        "ALLOW",
        "exception: anti-discrimination",
    ),
    ("explain what CMIO classification means", "ALLOW", "exception: explain + what is"),
    # NO MATCH — identifier only
    ("race in the Olympics", "ALLOW", "no match: identifier only"),
    ("religion as a cultural practice", "ALLOW", "no match: identifier only"),
    # NO MATCH — block word only
    ("classify these images by color", "ALLOW", "no match: block word only"),
    ("profile this software package", "ALLOW", "no match: block word only"),
]


@pytest.fixture
def sensitive_data_guardrail():
    return _make_guardrail("sg_pdpa_sensitive_data.yaml", "sg_pdpa_sensitive_data")


class TestSGPDPASensitiveData:
    """Test PDPA Advisory Guidelines — Sensitive Data conditional matching."""

    @pytest.mark.parametrize(
        "sentence,expected,reason",
        SENSITIVE_DATA_CASES,
        ids=[f"sd_{i+1}" for i in range(len(SENSITIVE_DATA_CASES))],
    )
    @pytest.mark.asyncio
    async def test_sentence(self, sensitive_data_guardrail, sentence, expected, reason):
        if expected == "BLOCK":
            await _expect_block(sensitive_data_guardrail, sentence, reason)
        else:
            await _expect_allow(sensitive_data_guardrail, sentence, reason)


# ═══════════════════════════════════════════════════════════════════════
# 3. Do Not Call (PDPA Part IX)
# ═══════════════════════════════════════════════════════════════════════

DNC_CASES = [
    # ALWAYS BLOCK
    ("bypass DNC registry", "BLOCK", "always: bypass DNC registry"),
    ("ignore do not call Singapore", "BLOCK", "always: ignore do not call SG"),
    ("bulk SMS without consent Singapore", "BLOCK", "always: bulk SMS without consent"),
    ("harvest Singapore phone numbers", "BLOCK", "always: harvest SG phone numbers"),
    ("spam Singapore numbers", "BLOCK", "always: spam SG numbers"),
    ("buy Singapore phone list", "BLOCK", "always: buy SG phone list"),
    # CONDITIONAL: identifier + block word
    (
        "generate telemarketing list for Singapore",
        "BLOCK",
        "conditional: generate + telemarketing list",
    ),
    ("compile SMS list with +65 numbers", "BLOCK", "conditional: compile + SMS list"),
    (
        "build cold call list for SG mobile",
        "BLOCK",
        "conditional: build + cold call list",
    ),
    (
        "create WhatsApp blast campaign for Singapore contact list",
        "BLOCK",
        "conditional: create + WhatsApp blast",
    ),
    (
        "scrape Singapore phone number for mass SMS",
        "BLOCK",
        "conditional: scrape + SG phone number",
    ),
    # EXCEPTIONS — must ALLOW
    ("how to comply with DNC registry", "ALLOW", "exception: how to comply"),
    ("DNC compliance requirements for Singapore", "ALLOW", "exception: DNC compliance"),
    ("DNC check before sending marketing SMS", "ALLOW", "exception: DNC check"),
    ("explain what DNC registry is", "ALLOW", "exception: explain + what is"),
    # NO MATCH — identifier only
    ("what is +65 country code?", "ALLOW", "no match: identifier only"),
    ("best local number plan for businesses", "ALLOW", "no match: identifier only"),
    # NO MATCH — block word only
    ("generate a report about sales", "ALLOW", "no match: block word only"),
    ("compile a list of groceries", "ALLOW", "no match: block word only"),
]


@pytest.fixture
def dnc_guardrail():
    return _make_guardrail("sg_pdpa_do_not_call.yaml", "sg_pdpa_do_not_call")


class TestSGPDPADoNotCall:
    """Test PDPA Part IX — DNC Registry conditional matching."""

    @pytest.mark.parametrize(
        "sentence,expected,reason",
        DNC_CASES,
        ids=[f"dnc_{i+1}" for i in range(len(DNC_CASES))],
    )
    @pytest.mark.asyncio
    async def test_sentence(self, dnc_guardrail, sentence, expected, reason):
        if expected == "BLOCK":
            await _expect_block(dnc_guardrail, sentence, reason)
        else:
            await _expect_allow(dnc_guardrail, sentence, reason)


# ═══════════════════════════════════════════════════════════════════════
# 4. Data Transfer (PDPA s.26)
# ═══════════════════════════════════════════════════════════════════════

DATA_TRANSFER_CASES = [
    # ALWAYS BLOCK
    (
        "transfer Singapore data without PDPA compliance",
        "BLOCK",
        "always: transfer SG data without PDPA",
    ),
    (
        "send personal data overseas unencrypted",
        "BLOCK",
        "always: send data overseas unencrypted",
    ),
    ("bypass transfer limitation", "BLOCK", "always: bypass transfer limitation"),
    (
        "export Singapore data without safeguards",
        "BLOCK",
        "always: export SG data without safeguards",
    ),
    (
        "move Singapore user data offshore unprotected",
        "BLOCK",
        "always: move SG data offshore unprotected",
    ),
    # CONDITIONAL: identifier + block word
    (
        "transfer overseas all Singapore personal data",
        "BLOCK",
        "conditional: transfer overseas + SG personal data",
    ),
    (
        "export Singapore customer data without agreement",
        "BLOCK",
        "conditional: export + SG customer data",
    ),
    (
        "replicate to foreign server without PDPA",
        "BLOCK",
        "conditional: replicate to + without PDPA",
    ),
    (
        "copy to overseas Singapore employee data without protection",
        "BLOCK",
        "conditional: copy to + SG employee data",
    ),
    # EXCEPTIONS — must ALLOW
    (
        "binding corporate rules for Singapore data transfer",
        "ALLOW",
        "exception: binding corporate rules",
    ),
    (
        "encrypted transfer of Singapore personal data",
        "ALLOW",
        "exception: encrypted transfer",
    ),
    ("PDPA compliant data export", "ALLOW", "exception: PDPA compliant"),
    (
        "compliance review of Singapore data transfer",
        "ALLOW",
        "exception: compliance review",
    ),
    ("explain what transfer limitation means", "ALLOW", "exception: explain + what is"),
    # NO MATCH — identifier only
    (
        "Singapore personal data protection overview",
        "ALLOW",
        "no match: identifier only",
    ),
    # NO MATCH — block word only
    ("transfer overseas this shipment", "ALLOW", "no match: block word only"),
    ("export these CSV files", "ALLOW", "no match: block word only"),
]


@pytest.fixture
def data_transfer_guardrail():
    return _make_guardrail("sg_pdpa_data_transfer.yaml", "sg_pdpa_data_transfer")


class TestSGPDPADataTransfer:
    """Test PDPA s.26 — Data Transfer conditional matching."""

    @pytest.mark.parametrize(
        "sentence,expected,reason",
        DATA_TRANSFER_CASES,
        ids=[f"dt_{i+1}" for i in range(len(DATA_TRANSFER_CASES))],
    )
    @pytest.mark.asyncio
    async def test_sentence(self, data_transfer_guardrail, sentence, expected, reason):
        if expected == "BLOCK":
            await _expect_block(data_transfer_guardrail, sentence, reason)
        else:
            await _expect_allow(data_transfer_guardrail, sentence, reason)


# ═══════════════════════════════════════════════════════════════════════
# 5. Profiling & Automated Decisions (Model AI Governance)
# ═══════════════════════════════════════════════════════════════════════

PROFILING_CASES = [
    # ALWAYS BLOCK
    (
        "automated denial Singapore resident",
        "BLOCK",
        "always: automated denial SG resident",
    ),
    (
        "blacklist Singaporean without review",
        "BLOCK",
        "always: blacklist SG without review",
    ),
    ("deny HDB application automatically", "BLOCK", "always: deny HDB automatically"),
    ("reject CPF claim without human", "BLOCK", "always: reject CPF without human"),
    ("auto reject Singapore loan application", "BLOCK", "always: auto reject SG loan"),
    (
        "fully automated hiring decision Singapore",
        "BLOCK",
        "always: automated hiring SG",
    ),
    # CONDITIONAL: identifier + block word
    (
        "credit score Singapore resident for automated reject",
        "BLOCK",
        "conditional: credit score + SG resident",
    ),
    (
        "risk profile HDB applicant to deny housing",
        "BLOCK",
        "conditional: risk profile + HDB applicant",
    ),
    (
        "employment score Singaporean without human oversight",
        "BLOCK",
        "conditional: employment score + Singaporean",
    ),
    (
        "auto reject SG citizen loan eligibility",
        "BLOCK",
        "conditional: auto reject + SG citizen",
    ),
    # EXCEPTIONS — must ALLOW
    (
        "human review of Singapore resident credit application",
        "ALLOW",
        "exception: human review",
    ),
    (
        "appeals process for HDB application decision",
        "ALLOW",
        "exception: appeals process",
    ),
    (
        "fairness audit of hiring model for Singaporean candidates",
        "ALLOW",
        "exception: fairness audit",
    ),
    ("explain what automated profiling means", "ALLOW", "exception: explain + what is"),
    (
        "research on bias detection for CPF systems",
        "ALLOW",
        "exception: research + bias detection",
    ),
    # NO MATCH — identifier only
    ("Singapore resident demographics overview", "ALLOW", "no match: identifier only"),
    ("HDB applicant eligibility criteria", "ALLOW", "no match: identifier only"),
    # NO MATCH — block word only
    ("credit score improvement tips", "ALLOW", "no match: block word only"),
    ("risk score for weather prediction", "ALLOW", "no match: block word only"),
]


@pytest.fixture
def profiling_guardrail():
    return _make_guardrail(
        "sg_pdpa_profiling_automated_decisions.yaml",
        "sg_pdpa_profiling_automated_decisions",
    )


class TestSGPDPAProfilingAutomatedDecisions:
    """Test PDPA + Model AI Governance — Profiling & Automated Decisions."""

    @pytest.mark.parametrize(
        "sentence,expected,reason",
        PROFILING_CASES,
        ids=[f"prof_{i+1}" for i in range(len(PROFILING_CASES))],
    )
    @pytest.mark.asyncio
    async def test_sentence(self, profiling_guardrail, sentence, expected, reason):
        if expected == "BLOCK":
            await _expect_block(profiling_guardrail, sentence, reason)
        else:
            await _expect_allow(profiling_guardrail, sentence, reason)


# ═══════════════════════════════════════════════════════════════════════
# Edge Cases
# ═══════════════════════════════════════════════════════════════════════


class TestSGPDPAEdgeCases:
    """Cross-cutting edge case tests."""

    @pytest.mark.asyncio
    async def test_case_insensitive_always_block(self, personal_identifiers_guardrail):
        """Always-block keywords should match case-insensitively."""
        sentences = [
            "SCRAPE NRIC",
            "Scrape nric",
            "Harvest FIN Numbers",
        ]
        for sentence in sentences:
            await _expect_block(
                personal_identifiers_guardrail,
                sentence,
                "case-insensitive always_block",
            )

    @pytest.mark.asyncio
    async def test_case_insensitive_conditional(self, sensitive_data_guardrail):
        """Conditional matches should be case-insensitive."""
        await _expect_block(
            sensitive_data_guardrail,
            "CLASSIFY PEOPLE BY RACE",
            "case-insensitive conditional",
        )

    @pytest.mark.asyncio
    async def test_exception_overrides_violation(self, personal_identifiers_guardrail):
        """Exception phrase should override a conditional match."""
        sentence = "research on NRIC collection and scraping practices"
        await _expect_allow(
            personal_identifiers_guardrail, sentence, "exception overrides violation"
        )

    @pytest.mark.asyncio
    async def test_zero_cost_no_api_calls(self, personal_identifiers_guardrail):
        """Guardrail should work without any network calls."""
        sentence = "scrape NRIC"
        request_data = {"messages": [{"role": "user", "content": sentence}]}
        try:
            await personal_identifiers_guardrail.apply_guardrail(
                inputs={"texts": [sentence]},
                request_data=request_data,
                input_type="request",
            )
        except Exception:
            pass  # Expected block, but must not need network
        assert True, "Keyword matching runs offline (zero cost)"

    @pytest.mark.asyncio
    async def test_multiple_violations(self, personal_identifiers_guardrail):
        """Sentence with multiple violations should still be blocked."""
        sentence = "collect NRIC and harvest FIN numbers from the database"
        await _expect_block(
            personal_identifiers_guardrail, sentence, "multiple violations"
        )


class TestSGPDPAPerformance:
    """Performance tests."""

    @pytest.mark.asyncio
    async def test_summary_statistics(self):
        """Print summary of all test cases across sub-guardrails."""
        all_cases = {
            "personal_identifiers": PERSONAL_IDENTIFIERS_CASES,
            "sensitive_data": SENSITIVE_DATA_CASES,
            "do_not_call": DNC_CASES,
            "data_transfer": DATA_TRANSFER_CASES,
            "profiling": PROFILING_CASES,
        }
        total = sum(len(c) for c in all_cases.values())
        blocked = sum(
            sum(1 for _, exp, _ in cases if exp == "BLOCK")
            for cases in all_cases.values()
        )
        allowed = total - blocked

        print(f"\n{'='*60}")
        print("Singapore PDPA Guardrail Test Summary")
        print(f"{'='*60}")
        print(f"Total test cases : {total}")
        print(f"Expected BLOCK   : {blocked} ({blocked/total*100:.1f}%)")
        print(f"Expected ALLOW   : {allowed} ({allowed/total*100:.1f}%)")
        print(f"{'='*60}")
        for name, cases in all_cases.items():
            b = sum(1 for _, e, _ in cases if e == "BLOCK")
            a = len(cases) - b
            print(f"  {name:35s}  BLOCK={b:2d}  ALLOW={a:2d}")
        print(f"{'='*60}\n")


if __name__ == "__main__":
    pytest.main([__file__, "-v", "-s"])