cpex-python 0.2.3

PyO3 bindings for the CPEX plugin runtime
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

# Location: ./bindings/python/tests/test_result.py
# Copyright 2025
# SPDX-License-Identifier: Apache-2.0
# Authors: Ted Habeck
#
# Tests for PyPipelineResult field access and repr (U5).

import re

import pytest
from cpex import PluginManager, PipelineResult


# ---------------------------------------------------------------------------
# Happy path field access
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_all_fields_accessible(manager: PluginManager):
    """All seven PipelineResult fields are accessible after a real invoke."""
    payload = {
        "message": {
            "role": "user",
            "content": [{"content_type": "text", "text": "Hello"}],
        }
    }
    result = await manager.invoke_hook("cmf.tool_pre_invoke", payload)
    assert isinstance(result, PipelineResult)

    # continue_processing
    assert isinstance(result.continue_processing, bool)
    # violation — None for an allowed result
    assert result.violation is None
    # errors — list (may be empty)
    assert isinstance(result.errors, list)
    # metadata — may be None
    assert result.metadata is None or isinstance(result.metadata, dict)
    # context_table — always a dict
    assert isinstance(result.context_table, dict)


# ---------------------------------------------------------------------------
# Deny result — violation dict shape
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_deny_result_violation_fields(pii_deny_config_path: str):
    """A denied result exposes violation dict with expected keys."""
    mgr = PluginManager(pii_deny_config_path)
    await mgr.initialize()

    payload = {
        "message": {
            "role": "assistant",
            "content": [
                {
                    "content_type": "tool_call",
                    "content": {
                        "tool_call_id": "tc_x",
                        "name": "submit_form",
                        "arguments": {"ssn": "987-65-4321"},
                    },
                }
            ],
        }
    }
    result = await mgr.invoke_hook("cmf.tool_pre_invoke", payload)

    assert result.continue_processing is False
    assert result.violation is not None
    assert isinstance(result.violation, dict)
    # Standard violation keys
    assert "reason" in result.violation

    await mgr.shutdown()


# ---------------------------------------------------------------------------
# repr safety — no pointer leakage (R3)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_repr_no_pointers(manager: PluginManager):
    """__repr__ must not contain hex pointer substrings (R3)."""
    payload = {
        "message": {
            "role": "user",
            "content": [{"content_type": "text", "text": "hi"}],
        }
    }
    result = await manager.invoke_hook("cmf.tool_pre_invoke", payload)
    r = repr(result)
    # Hex-pointer pattern: 0x followed by hex digits
    assert not re.search(r"0x[0-9a-fA-F]+", r), f"repr contains pointer: {r!r}"