agent-memory-api/test_eligibility.py at main · sasuke15134321/agent-memory-api · GitHub

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
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
"""
Usage Eligibility Atom v0.1 - Local Tests
No network, no database, no external dependencies.
"""
from eligibility_engine import (
    evaluate_usage_eligibility,
    MemoryRecord,
    RequestContext,
    PolicySnapshot,
    LifecycleSnapshot,
    build_eligibility_input,
)

POLICY_VERSION = "usage-policy-v0.1"
BASE = dict(
    memory_id="mem_test_001",
    subject_id="customer-123",
    current_subject_id="customer-123",
    purpose_scope=["customer_support", "order_status"],
    requested_purpose="customer_support",
    lifecycle_status="active",
    revocation_status="not_revoked",
    policy_version=POLICY_VERSION,
)


def test_allowed():
    r = evaluate_usage_eligibility(**BASE)
    assert r.decision == "allowed"
    assert r.reason_code == "ELIGIBILITY_REQUIREMENTS_SATISFIED"


def test_purpose_mismatch():
    r = evaluate_usage_eligibility(**{**BASE, "requested_purpose": "payment_decision"})
    assert r.decision == "denied"
    assert r.reason_code == "PURPOSE_SCOPE_MISMATCH"


def test_subject_mismatch():
    r = evaluate_usage_eligibility(**{**BASE, "current_subject_id": "customer-999"})
    assert r.decision == "denied"
    assert r.reason_code == "SUBJECT_SCOPE_MISMATCH"


def test_inactive():
    r = evaluate_usage_eligibility(**{**BASE, "lifecycle_status": "expired"})
    assert r.decision == "denied"
    assert r.reason_code == "INFORMATION_NOT_ACTIVE"


def test_revoked():
    r = evaluate_usage_eligibility(**{**BASE, "revocation_status": "revoked"})
    assert r.decision == "denied"
    assert r.reason_code == "INFORMATION_REVOKED"


def test_unknown_revocation_status():
    r = evaluate_usage_eligibility(**{**BASE, "revocation_status": "unknown"})
    assert r.decision == "unknown"
    assert r.reason_code == "REVOCATION_STATUS_UNAVAILABLE"


def test_undefined_revocation_status():
    r = evaluate_usage_eligibility(**{**BASE, "revocation_status": "pending"})
    assert r.decision == "unknown"
    assert r.reason_code == "REVOCATION_STATUS_UNAVAILABLE"


def test_missing_required_field():
    r = evaluate_usage_eligibility(**{**BASE, "memory_id": None})
    assert r.decision == "unknown"
    assert r.reason_code == "MISSING_REQUIRED_FIELD"


# --- Shadow Adapter tests (max 3) ---

def test_shadow_adapter_allowed_full_snapshots():
    mem = MemoryRecord(memory_id="mem_abc123", agent_id="memory-writer-agent")
    mem_id_before = mem.memory_id
    agent_id_before = mem.agent_id

    req = RequestContext(
        current_subject_id="customer-123",
        requested_purpose="customer_support",
        requesting_agent_id="support-agent-7f3a",
    )
    pol = PolicySnapshot(
        policy_version="usage-policy-v0.1",
        purpose_scope=["customer_support", "order_status"],
    )
    lc = LifecycleSnapshot(
        subject_id="customer-123",
        lifecycle_status="active",
        revocation_status="not_revoked",
    )

    r = build_eligibility_input(mem, req, pol, lc)

    assert r.decision == "allowed"
    assert r.reason_code == "ELIGIBILITY_REQUIREMENTS_SATISFIED"
    # Memory object must not be mutated
    assert mem.memory_id == mem_id_before
    assert mem.agent_id == agent_id_before


def test_shadow_adapter_purpose_mismatch():
    mem = MemoryRecord(memory_id="mem_abc123", agent_id="memory-writer-agent")
    req = RequestContext(
        current_subject_id="customer-123",
        requested_purpose="payment_decision",
    )
    pol = PolicySnapshot(
        policy_version="usage-policy-v0.1",
        purpose_scope=["customer_support", "order_status"],
    )
    lc = LifecycleSnapshot(
        subject_id="customer-123",
        lifecycle_status="active",
        revocation_status="not_revoked",
    )

    r = build_eligibility_input(mem, req, pol, lc)

    assert r.decision == "denied"
    assert r.reason_code == "PURPOSE_SCOPE_MISMATCH"


def test_shadow_adapter_missing_external_state():
    mem = MemoryRecord(memory_id="mem_abc123", agent_id="memory-writer-agent")
    req = RequestContext(
        current_subject_id="customer-123",
        requested_purpose="customer_support",
    )
    pol = PolicySnapshot(
        policy_version="usage-policy-v0.1",
        purpose_scope=["customer_support", "order_status"],
    )
    # revocation_status is missing (None)
    lc = LifecycleSnapshot(
        subject_id="customer-123",
        lifecycle_status="active",
        revocation_status=None,
    )

    r = build_eligibility_input(mem, req, pol, lc)

    assert r.decision == "unknown"
    assert r.reason_code == "MISSING_REQUIRED_FIELD"