34a3123799
Read the resource-control open/restricted signal and refuse new orders that would exceed a restricted entity's published allowance. Open and missing signals stay unchanged. Safety paths admit with an exception.
139 lines
4.5 KiB
Python
139 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
|
|
REPO_DIR = Path(__file__).resolve().parents[1]
|
|
SPEC = importlib.util.spec_from_file_location(
|
|
"consumption_mode", REPO_DIR / "scripts/consumption_mode.py"
|
|
)
|
|
cm = importlib.util.module_from_spec(SPEC)
|
|
assert SPEC.loader is not None
|
|
sys.modules[SPEC.name] = cm
|
|
SPEC.loader.exec_module(cm)
|
|
|
|
|
|
def restricted(allowance: str = "45.83") -> dict:
|
|
return {
|
|
"schema_version": "0.1",
|
|
"record_type": "consumption_mode",
|
|
"financial_entity_id": "entity:coulomb",
|
|
"period": "2026-09",
|
|
"consumption_mode": "restricted",
|
|
"new_transfer_charges_allowed_eur": allowance,
|
|
"terms_version": "0.1",
|
|
}
|
|
|
|
|
|
class ConsumptionModeTests(unittest.TestCase):
|
|
def test_open_entity_unchanged(self) -> None:
|
|
signal = restricted()
|
|
signal["consumption_mode"] = "open"
|
|
signal["new_transfer_charges_allowed_eur"] = None
|
|
result = cm.decide(
|
|
entity_id="entity:coulomb",
|
|
order_class="new-order",
|
|
estimate_eur=Decimal("200.00"),
|
|
signal=signal,
|
|
)
|
|
self.assertEqual(result["decision"], "admit")
|
|
self.assertEqual(result["reason"], "open")
|
|
|
|
def test_restricted_cannot_exceed_allowance(self) -> None:
|
|
result = cm.decide(
|
|
entity_id="entity:coulomb",
|
|
order_class="new-order",
|
|
estimate_eur=Decimal("45.84"),
|
|
signal=restricted(),
|
|
)
|
|
self.assertEqual(result["decision"], "refuse")
|
|
self.assertEqual(result["reason"], "exceeds-allowance")
|
|
|
|
def test_restricted_within_allowance(self) -> None:
|
|
result = cm.decide(
|
|
entity_id="entity:coulomb",
|
|
order_class="new-order",
|
|
estimate_eur=Decimal("45.83"),
|
|
signal=restricted(),
|
|
)
|
|
self.assertEqual(result["decision"], "admit")
|
|
self.assertEqual(result["reason"], "within-allowance")
|
|
|
|
def test_restricted_without_estimate_refused(self) -> None:
|
|
result = cm.decide(
|
|
entity_id="entity:coulomb",
|
|
order_class="elastic",
|
|
estimate_eur=None,
|
|
signal=restricted(),
|
|
)
|
|
self.assertEqual(result["decision"], "refuse")
|
|
self.assertEqual(result["reason"], "restricted-estimate-required")
|
|
|
|
def test_safety_admitted_as_exception(self) -> None:
|
|
result = cm.decide(
|
|
entity_id="entity:coulomb",
|
|
order_class="safety",
|
|
estimate_eur=Decimal("200.00"),
|
|
signal=restricted(),
|
|
)
|
|
self.assertEqual(result["decision"], "admit")
|
|
self.assertTrue(result["exception"])
|
|
|
|
def test_no_signal_is_not_restricted(self) -> None:
|
|
result = cm.decide(
|
|
entity_id="entity:coulomb",
|
|
order_class="new-order",
|
|
estimate_eur=Decimal("999.00"),
|
|
signal=None,
|
|
)
|
|
self.assertEqual(result["decision"], "admit")
|
|
self.assertEqual(result["reason"], "no-signal-not-restricted")
|
|
|
|
def test_railiance_self_use_always_open(self) -> None:
|
|
result = cm.decide(
|
|
entity_id="entity:railiance",
|
|
order_class="new-order",
|
|
estimate_eur=Decimal("999.00"),
|
|
signal=restricted(),
|
|
)
|
|
self.assertEqual(result["decision"], "admit")
|
|
self.assertEqual(result["reason"], "railiance-self-use")
|
|
|
|
def test_cli_refuses_restricted_overage(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / "signal.json"
|
|
path.write_text(json.dumps([restricted()]), encoding="utf-8")
|
|
code = cm.main(
|
|
[
|
|
"check",
|
|
"--entity",
|
|
"entity:coulomb",
|
|
"--estimate-eur",
|
|
"50",
|
|
"--signal",
|
|
str(path),
|
|
]
|
|
)
|
|
self.assertEqual(code, 2)
|
|
|
|
def test_cli_admits_open(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / "signal.json"
|
|
path.write_text("[]", encoding="utf-8")
|
|
code = cm.main(
|
|
[
|
|
"check",
|
|
"--entity",
|
|
"entity:coulomb",
|
|
"--estimate-eur",
|
|
"50",
|
|
"--signal",
|
|
str(path),
|
|
]
|
|
)
|
|
self.assertEqual(code, 0)
|