Diameter¶

The diameter namespace exposes the IMS Diameter interfaces — Cx (HSS), Rx (PCRF), Sh (HSS AS), and Rf (offline charging) — plus a unified inbound @diameter.on_request hook for serving requests (RAR, PNR, ASR, …).

from siphon import diameter

@diameter.on_request
async def handle(request):
    if request.command_name == "RAR":
        return request.answer(2001)
    return request.reject(3002)

`diameter` namespace¶

Mock Diameter namespace for testing scripts that use from siphon import diameter.

Exposes connection status and Cx/Rx methods matching the Rust DiameterNamespace.

Example::

from siphon_sdk import mock_module
mock_module.install()
diameter = mock_module.get_diameter()
diameter.add_peer("hss1", connected=True)
diameter.set_default_server_name("sip:scscf.ims.example.com:6060")

from siphon import diameter
assert diameter.is_connected("hss1")
result = diameter.cx_uar("sip:alice@ims.example.com")
assert result["server_name"] == "sip:scscf.ims.example.com:6060"

config `property` ¶

config: dict

Read-only view of the parsed diameter config (tenants/listen).

Set it in tests with diameter.set_config({...}).

event_sink `property` ¶

event_sink: 'MockEventSink'

The generic event sink (diameter.event_sink.emit(row)).

is_connected ¶

is_connected(peer_name: str) -> bool

Check if a Diameter peer is connected.

Parameters:

Name	Type	Description	Default
`peer_name`	`str`	Name of the peer (e.g. "hss1").	required

Returns:

Type	Description
`bool`	`True` if the peer was added and is marked as connected.

peer_count ¶

peer_count() -> int

Get the number of connected peers.

Returns:

Type	Description
`int`	Count of peers that are marked as connected.

decode_isdn_address ¶

decode_isdn_address(value: Union[bytes, str]) -> str

Decode an ISDN-AddressString to its E.164 digit string.

Accepts the raw AVP bytes (0x91 ToN/NPI + TBCD digits) or an already-decoded str — the latter is returned unchanged, so it is safe to call on the result of req.get_avp("MSISDN") regardless of the AVP's dictionary type. A missing ToN/NPI byte is tolerated.

Parameters:

Name	Type	Description	Default
`value`	`Union[bytes, str]`	`bytes` (raw ISDN-AddressString) or `str` (digits).	required

Returns:

Type	Description
`str`	The E.164 digit string (no leading `+`).

Example

diameter.decode_isdn_address(req.get_avp("MSISDN")) '31612345678'

encode_isdn_address ¶

encode_isdn_address(
    digits: str, ton_npi: int = TON_NPI_INTERNATIONAL_E164
) -> bytes

Encode an E.164 digit string as an ISDN-AddressString — one ToN/NPI octet followed by the TBCD digit string.

Use when building a raw OctetString AVP by hand for an unknown code; dictionary-typed AVPs (MSISDN / SC-Address / SGSN-Number / MME-Number-for-MT-SMS) encode digit strings automatically. A leading + is stripped.

Parameters:

Name	Type	Description	Default
`digits`	`str`	The E.164 number as a digit string.	required
`ton_npi`	`int`	ToN/NPI byte (default `0x91` = international E.164).	`TON_NPI_INTERNATIONAL_E164`

Returns:

Type	Description
`bytes`	The encoded ISDN-AddressString `bytes`.

Example

diameter.encode_isdn_address("31612345678") b'\x91\x13\x16\x32\x54\x76\xf8'

cx_uar ¶

cx_uar(
    public_identity: str,
    visited_network_id: Optional[str] = None,
    user_auth_type: Optional[int] = None,
) -> Optional[dict]

Send a User-Authorization-Request to discover S-CSCF assignment.

Parameters:

Name	Type	Description	Default
`public_identity`	`str`	User's public identity (e.g. `"sip:alice@ims.example.com"`).	required
`visited_network_id`	`Optional[str]`	Visited network identifier.	`None`
`user_auth_type`	`Optional[int]`	User-Authorization-Type AVP value (3GPP TS 29.229). `0` = REGISTRATION, `1` = DE_REGISTRATION, `2` = REGISTRATION_AND_CAPABILITIES. Omit to not send the AVP.	`None`

Returns:

Type	Description
`Optional[dict]`	Dict with `result_code` and `server_name`, or `None`.

cx_sar ¶

cx_sar(
    public_identity: str,
    server_name: Optional[str] = None,
    assignment_type: int = 1,
) -> Optional[dict]

Send a Server-Assignment-Request after REGISTER auth.

Parameters:

Name	Type	Description	Default
`public_identity`	`str`	User's public identity.	required
`server_name`	`Optional[str]`	This S-CSCF's SIP URI.	`None`
`assignment_type`	`int`	Server-Assignment-Type (default 1 = REGISTRATION).	`1`

Returns:

Type	Description
`Optional[dict]`	Dict with `result_code` and `user_data` (iFC XML), or `None`.

cx_lir ¶

cx_lir(public_identity: str) -> Optional[dict]

Send a Location-Info-Request to find the serving S-CSCF.

Parameters:

Name	Type	Description	Default
`public_identity`	`str`	Target user's public identity.	required

Returns:

Type	Description
`Optional[dict]`	Dict with `result_code` and `server_name`, or `None`.

rx_aar ¶

rx_aar(
    session_id: Optional[str] = None,
    framed_ip: Optional[str] = None,
    framed_ipv6: Union[str, bytes, None] = None,
    media_components: Optional[list] = None,
    af_application_id: str = "IMS Services",
    subscription_id: Optional[tuple] = None,
) -> Optional[dict]

Send an Rx AA-Request for QoS resource reservation.

Parameters:

Name	Type	Description	Default
`session_id`	`Optional[str]`	Reuse an existing Rx session ID (modification AAR per TS 29.214 §4.4.5). `None` allocates a new session.	`None`
`framed_ip`	`Optional[str]`	UE IPv4 address (Framed-IP-Address AVP).	`None`
`framed_ipv6`	`Union[str, bytes, None]`	UE IPv6 address (str or bytes).	`None`
`media_components`	`Optional[list]`	list of media-component dicts shaped per TS 29.214 §5.3.7 (see project docs for the full schema).	`None`
`af_application_id`	`str`	AF-Application-Identifier (default `"IMS Services"`).	`'IMS Services'`
`subscription_id`	`Optional[tuple]`	Optional `(data, type)` tuple identifying the IMS subscriber (RFC 4006 §8.47).	`None`

Returns:

Type	Description
`Optional[dict]`	Dict with `result_code` and `session_id`, or `None`.

rx_str ¶

rx_str(session_id: str) -> Optional[int]

Send an Rx Session-Termination-Request.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The Rx session ID from the original AAR.	required

Returns:

Type	Description
`Optional[int]`	Result code (int), or `None`.

sh_udr ¶

sh_udr(
    public_identity: str,
    data_reference: Union[int, list[int]],
    service_indication: Optional[str] = None,
) -> Optional[dict]

Send a Sh User-Data-Request to fetch user profile data from the HSS.

Parameters:

Name	Type	Description	Default
`public_identity`	`str`	Target user's public identity.	required
`data_reference`	`Union[int, list[int]]`	Data-Reference int or list[int] (TS 29.328 §7.6).	required
`service_indication`	`Optional[str]`	e.g. `"simservs"` for Repository-Data.	`None`

Returns:

Type	Description
`Optional[dict]`	Dict with `result_code` and `user_data` (XML), or `None`.

sh_pur ¶

sh_pur(
    public_identity: str,
    data_reference: int,
    xml: str,
    service_indication: Optional[str] = None,
) -> Optional[dict]

Send a Sh Profile-Update-Request to push user profile data to the HSS.

Parameters:

Name	Type	Description	Default
`public_identity`	`str`	Target user's public identity.	required
`data_reference`	`int`	Data-Reference (e.g. `0` = Repository-Data).	required
`xml`	`str`	UTF-8 XML payload.	required
`service_indication`	`Optional[str]`	e.g. `"simservs"`; required by the HSS when Data-Reference is Repository-Data (TS 29.328 §6.1.3).	`None`

Returns:

Type	Description
`Optional[dict]`	Dict with `result_code`, or `None`.

sh_snr ¶

sh_snr(
    public_identity: str,
    data_reference: Union[int, list[int]],
    subs_req_type: int,
    service_indication: Optional[str] = None,
) -> Optional[dict]

Send a Sh Subscribe-Notifications-Request to the HSS.

Parameters:

Name	Type	Description	Default
`public_identity`	`str`	Target user's public identity.	required
`data_reference`	`Union[int, list[int]]`	Data-Reference int or list[int] to subscribe to.	required
`subs_req_type`	`int`	`0` = SUBSCRIBE, `1` = UNSUBSCRIBE.	required
`service_indication`	`Optional[str]`	e.g. `"simservs"`; required by the HSS when Data-Reference is Repository-Data (TS 29.328 §6.1.4).	`None`

Returns:

Type	Description
`Optional[dict]`	Dict with `result_code`, or `None`.

add_peer ¶

add_peer(name: str, connected: bool = True) -> None

Register a mock Diameter peer (test helper).

Parameters:

Name	Type	Description	Default
`name`	`str`	Peer name.	required
`connected`	`bool`	Whether the peer should appear as connected.	`True`

set_default_server_name ¶

set_default_server_name(server_name: str) -> None

Set a default S-CSCF name returned by UAR/LIR when no per-user response is configured.

Parameters:

Name	Type	Description	Default
`server_name`	`str`	S-CSCF SIP URI (e.g. `"sip:scscf.ims.example.com:6060"`).	required

set_uar_response ¶

set_uar_response(
    public_identity: str,
    result_code: int = 2001,
    server_name: Optional[str] = None,
) -> None

Configure a mock UAA response for a specific user (test helper).

Parameters:

Name	Type	Description	Default
`public_identity`	`str`	User's public identity.	required
`result_code`	`int`	Diameter result code (default 2001 = SUCCESS).	`2001`
`server_name`	`Optional[str]`	Assigned S-CSCF URI.	`None`

set_sar_response ¶

set_sar_response(
    public_identity: str,
    result_code: int = 2001,
    user_data: Optional[str] = None,
) -> None

Configure a mock SAA response for a specific user (test helper).

Parameters:

Name	Type	Description	Default
`public_identity`	`str`	User's public identity.	required
`result_code`	`int`	Diameter result code.	`2001`
`user_data`	`Optional[str]`	iFC XML string from user profile.	`None`

set_lir_response ¶

set_lir_response(
    public_identity: str,
    result_code: int = 2001,
    server_name: Optional[str] = None,
) -> None

Configure a mock LIA response for a specific user (test helper).

Parameters:

Name	Type	Description	Default
`public_identity`	`str`	User's public identity.	required
`result_code`	`int`	Diameter result code.	`2001`
`server_name`	`Optional[str]`	Serving S-CSCF URI.	`None`

set_aar_response ¶

set_aar_response(
    session_id: str, result_code: int = 2001
) -> None

Configure a mock AAA response for a specific Rx session (test helper).

Parameters:

Name	Type	Description	Default
`session_id`	`str`	Rx session ID.	required
`result_code`	`int`	Diameter result code.	`2001`

rf_acr_start ¶

rf_acr_start(
    *,
    calling_party: Optional[str] = None,
    called_party: Optional[str] = None,
    sip_method: Optional[str] = None,
    role_of_node: Optional[str] = None,
    node_functionality: Optional[str] = None,
    ims_charging_identifier: Optional[str] = None,
    user_session_id: Optional[str] = None,
    originating_ioi: Optional[str] = None,
    terminating_ioi: Optional[str] = None,
    application_server: Optional[str] = None,
    application_provided_called_party_address: Optional[
        str
    ] = None,
    incoming_trunk_group_id: Optional[str] = None,
    outgoing_trunk_group_id: Optional[str] = None,
    visited_network_id: Optional[str] = None,
    user_name: Optional[str] = None,
    cause_code: Optional[int] = None,
    service_context_id: Optional[str] = None,
    peer: Optional[str] = None
) -> Optional[dict]

Send Rf ACR-START to the CDF (TS 32.299 §6.2.2).

rf_acr_interim ¶

rf_acr_interim(
    session_id: str,
    record_number: int,
    *,
    calling_party: Optional[str] = None,
    called_party: Optional[str] = None,
    sip_method: Optional[str] = None,
    role_of_node: Optional[str] = None,
    node_functionality: Optional[str] = None,
    ims_charging_identifier: Optional[str] = None,
    user_session_id: Optional[str] = None,
    originating_ioi: Optional[str] = None,
    terminating_ioi: Optional[str] = None,
    application_server: Optional[str] = None,
    application_provided_called_party_address: Optional[
        str
    ] = None,
    incoming_trunk_group_id: Optional[str] = None,
    outgoing_trunk_group_id: Optional[str] = None,
    visited_network_id: Optional[str] = None,
    user_name: Optional[str] = None,
    cause_code: Optional[int] = None,
    service_context_id: Optional[str] = None,
    peer: Optional[str] = None
) -> Optional[dict]

Send Rf ACR-INTERIM (mid-session accounting update).

rf_acr_stop ¶

rf_acr_stop(
    session_id: str,
    record_number: int,
    *,
    termination_cause: int = 1,
    calling_party: Optional[str] = None,
    called_party: Optional[str] = None,
    sip_method: Optional[str] = None,
    role_of_node: Optional[str] = None,
    node_functionality: Optional[str] = None,
    ims_charging_identifier: Optional[str] = None,
    user_session_id: Optional[str] = None,
    originating_ioi: Optional[str] = None,
    terminating_ioi: Optional[str] = None,
    application_server: Optional[str] = None,
    application_provided_called_party_address: Optional[
        str
    ] = None,
    incoming_trunk_group_id: Optional[str] = None,
    outgoing_trunk_group_id: Optional[str] = None,
    visited_network_id: Optional[str] = None,
    user_name: Optional[str] = None,
    cause_code: Optional[int] = None,
    service_context_id: Optional[str] = None,
    peer: Optional[str] = None
) -> Optional[dict]

Send Rf ACR-STOP. termination_cause per RFC 6733 §8.15 (1=LOGOUT, 4=ADMINISTRATIVE, 5=LINK_BROKEN, 8=SESSION_TIMEOUT).

rf_acr_event ¶

rf_acr_event(
    *,
    calling_party: Optional[str] = None,
    called_party: Optional[str] = None,
    sip_method: Optional[str] = None,
    role_of_node: Optional[str] = None,
    node_functionality: Optional[str] = None,
    ims_charging_identifier: Optional[str] = None,
    user_session_id: Optional[str] = None,
    originating_ioi: Optional[str] = None,
    terminating_ioi: Optional[str] = None,
    application_server: Optional[str] = None,
    application_provided_called_party_address: Optional[
        str
    ] = None,
    incoming_trunk_group_id: Optional[str] = None,
    outgoing_trunk_group_id: Optional[str] = None,
    visited_network_id: Optional[str] = None,
    user_name: Optional[str] = None,
    cause_code: Optional[int] = None,
    service_context_id: Optional[str] = None,
    peer: Optional[str] = None
) -> Optional[dict]

Send Rf ACR-EVENT (one-shot accounting — REGISTER/MESSAGE).

set_rf_result_code ¶

set_rf_result_code(code: int) -> None

Override the Result-Code returned by every Rf ACA (default 2001).

set_rf_interim_interval ¶

set_rf_interim_interval(
    interval_secs: Optional[int],
) -> None

Configure the Acct-Interim-Interval returned in ACA-START.

captured_acrs ¶

captured_acrs() -> list[dict]

Return all ACRs the script has emitted via rf_acr_*.

Returns a fresh copy on each call. Useful for asserting on accounting flows in tests.

clear_captured_acrs ¶

clear_captured_acrs() -> None

Reset the captured-ACR list between tests.

ro_ccr_initial `async` ¶

ro_ccr_initial(
    subscription_id: str,
    *,
    subscription_id_type: Optional[str] = None,
    service_context_id: Optional[str] = None,
    requested_seconds: Optional[int] = None,
    rating_group: Optional[int] = None,
    service_identifier: Optional[int] = None,
    calling_party: Optional[str] = None,
    called_party: Optional[str] = None,
    sip_method: Optional[str] = None,
    role_of_node: Optional[str] = None,
    node_functionality: Optional[str] = None,
    ims_charging_identifier: Optional[str] = None,
    user_session_id: Optional[str] = None,
    originating_ioi: Optional[str] = None,
    terminating_ioi: Optional[str] = None,
    application_server: Optional[str] = None,
    application_provided_called_party_address: Optional[
        str
    ] = None,
    incoming_trunk_group_id: Optional[str] = None,
    outgoing_trunk_group_id: Optional[str] = None,
    visited_network_id: Optional[str] = None,
    cause_code: Optional[int] = None,
    peer: Optional[str] = None
) -> Optional[dict]

Send a Ro CCR-INITIAL and return the CCA dict.

Returns {result_code, session_id, request_number, granted_time, validity_time, final_unit_action}. For SCUR, thread the returned session_id through :meth:ro_ccr_update / :meth:ro_ccr_terminate.

Example

answer = await diameter.ro_ccr_initial( "+310000000001", requested_seconds=30, rating_group=100, calling_party="sip:alice@ims", called_party="sip:bob@ims") if answer["result_code"] != 2001: call.reject(402, "Payment Required")

ro_ccr_update `async` ¶

ro_ccr_update(
    subscription_id: str,
    session_id: str,
    request_number: int,
    *,
    subscription_id_type: Optional[str] = None,
    service_context_id: Optional[str] = None,
    used_seconds: Optional[int] = None,
    requested_seconds: Optional[int] = None,
    rating_group: Optional[int] = None,
    service_identifier: Optional[int] = None,
    peer: Optional[str] = None
) -> Optional[dict]

Send a Ro CCR-UPDATE reporting usage and requesting the next quota.

ro_ccr_terminate `async` ¶

ro_ccr_terminate(
    subscription_id: str,
    session_id: str,
    request_number: int,
    *,
    subscription_id_type: Optional[str] = None,
    service_context_id: Optional[str] = None,
    used_seconds: Optional[int] = None,
    rating_group: Optional[int] = None,
    service_identifier: Optional[int] = None,
    peer: Optional[str] = None
) -> Optional[dict]

Send a Ro CCR-TERMINATION closing the session with final usage.

ro_ccr_event `async` ¶

ro_ccr_event(
    subscription_id: str,
    *,
    subscription_id_type: Optional[str] = None,
    service_context_id: Optional[str] = None,
    requested_action: Optional[int] = None,
    calling_party: Optional[str] = None,
    called_party: Optional[str] = None,
    node_functionality: Optional[str] = None,
    user_session_id: Optional[str] = None,
    originator_address: Optional[str] = None,
    recipient_address: Optional[str] = None,
    sm_message_type: Optional[int] = None,
    sm_service_type: Optional[int] = None,
    sms_node: Optional[int] = None,
    data_coding_scheme: Optional[int] = None,
    peer: Optional[str] = None
) -> Optional[dict]

Send a one-shot Ro CCR-EVENT (IEC — SMS/RCS DIRECT_DEBITING).

Example

answer = await diameter.ro_ccr_event( "+310000000001", service_context_id="32274@3gpp.org", originator_address="+310000000001", recipient_address="+310000000002", sm_message_type=0) if answer["result_code"] != 2001: request.reply(402, "Payment Required") # no balance

set_ro_result_code ¶

set_ro_result_code(code: int) -> None

Override the Result-Code returned by every Ro CCA (default 2001).

set_ro_granted_time ¶

set_ro_granted_time(seconds: Optional[int]) -> None

Configure the granted CC-Time (seconds) returned in a successful CCA.

set_ro_final_unit_action ¶

set_ro_final_unit_action(action: Optional[int]) -> None

Configure the Final-Unit-Action (0=TERMINATE) returned in the CCA.

captured_ccrs ¶

captured_ccrs() -> list[dict]

Return all CCRs the script has emitted via ro_ccr_* (fresh copy).

clear_captured_ccrs ¶

clear_captured_ccrs() -> None

Reset the captured-CCR list between tests.

on_inbound_cer `staticmethod` ¶

on_inbound_cer(fn: Any) -> Any

Register the server-mode CER identity callback.

Called for an already-authenticated peer (both Rust auth gates have passed) with (peer_addr, peer_name, asserted_origin_host). Return (origin_host, origin_realm) to accept, or None to reject.

Example::

@diameter.on_inbound_cer
def cer_received(peer_addr, peer_name, asserted_origin_host):
    identity = diameter.config["tenants"]["default"]["identity"]
    return identity["origin_host"], identity["origin_realm"]

on_request `staticmethod` ¶

on_request(arg: Any = None) -> Any

Register the server-mode inbound-request dispatcher.

Called for inbound requests (R-bit set). Return req.reject(code), await req.forward_to(peer, ...), req.answer(code), or None (→ DIAMETER_UNABLE_TO_DELIVER, 3002).

An optional command filter scopes the handler (mirrors @proxy.on_request("INVITE")): bare @diameter.on_request (all), @diameter.on_request("ULR"), "ULR|AIR", or app-qualified "S6a:ULR". The mock treats it as an identity decorator either way.

Example::

@diameter.on_request("S6a:ULR")
async def update_location(req):
    return req.answer(2001)

on_reply `staticmethod` ¶

on_reply(fn: Any) -> Any

Register the server-mode answer-rewrite hook.

Called with (req, answer) on the answer an on_request handler produced — relayed via forward_to or built by answer/reject — just before it goes back upstream. A central place to rewrite answer AVPs for every reply (topology hiding, Origin-Host/Result-Code mapping). Mutate answer in place; the return value is ignored.

on_request_completed `staticmethod` ¶

on_request_completed(fn: Any) -> Any

Register the server-mode post-answer hook.

Called after the answer is sent upstream with (req, answer, latency_us) — typically to emit an event.

peer_pool ¶

peer_pool(
    target: Any, tenant: str = "default"
) -> "MockPeerPool"

Build a mock backend peer pool. target is a peer name or list of names; tenant is an optional scope label (defaults to "default" — single-domain servers leave it unset). Register backends with :meth:add_peer(connected=True).

ip_in_cidr `staticmethod` ¶

ip_in_cidr(addr: str, cidr: str) -> bool

Whether addr falls within cidr (mirrors the Rust helper).

fnmatch `staticmethod` ¶

fnmatch(value: str, pattern: str) -> bool

Shell-style glob match (*/?).

now_us `staticmethod` ¶

now_us() -> int

Wall-clock microseconds since the Unix epoch.

set_config ¶

set_config(config: dict) -> None

Test helper: set the dict returned by :attr:config.

s6a_air ¶

s6a_air(
    imsi: str,
    visited_plmn_id: bytes,
    num_vectors: int = 1,
    immediate_response_preferred: bool = True,
    resync_info: Optional[bytes] = None,
    peer: Optional[str] = None,
) -> Optional[dict]

Mock Authentication-Information. Returns canned E-UTRAN vectors; configure with :meth:set_air_response.

s6a_ulr ¶

s6a_ulr(
    imsi: str,
    visited_plmn_id: bytes,
    rat_type: int = 1004,
    ulr_flags: int = 0,
    peer: Optional[str] = None,
) -> Optional[dict]

Mock Update-Location. Returns a 2001 with subscription data present.

s6a_purge_ue ¶

s6a_purge_ue(
    imsi: str,
    pur_flags: Optional[int] = None,
    peer: Optional[str] = None,
) -> Optional[dict]

Mock Purge-UE. Returns a 2001.

s6c_srr ¶

s6c_srr(
    msisdn: str,
    sc_address: str,
    sm_rp_mti: Optional[int] = None,
) -> Optional[dict]

Mock Send-Routing-Info-for-SM. Configure responses via :meth:set_srr_response; default is a successful answer with an empty served-node (test scripts can detect the unset case).

s6c_rsr ¶

s6c_rsr(
    user_name: str, sc_address: str, delivery_outcome: int
) -> Optional[dict]

Mock Report-SM-Delivery-Status. Records the call on self.rsrs for assertions and returns a 2001.

sgd_tfr ¶

sgd_tfr(
    user_name: str,
    sc_address: str,
    sm_rp_ui: bytes,
    smsmi_correlation_id: Optional[str] = None,
    sm_rp_mti: Optional[int] = None,
) -> Optional[dict]

Mock MT-Forward-Short-Message. Records the TPDU on self.tfrs for assertions; returns 2001 unless overridden via :meth:set_tfr_response.

send_request ¶

send_request(
    command: str,
    application: str,
    peer: Optional[str] = None,
    timeout_ms: int = 10000,
    **avps: Any
) -> Optional[dict]

Generic Diameter request by spec name.

Records every call on self.generic_requests for assertions. Returns a default 2001-success answer unless overridden via :meth:set_generic_response.

set_generic_response ¶

set_generic_response(
    command: str, application: str, **answer: Any
) -> None

Configure a mock answer for send_request(command, application, ...).

set_udr_response ¶

set_udr_response(
    public_identity: str,
    result_code: int = 2001,
    user_data: Optional[str] = None,
) -> None

Configure a mock UDA response for a specific user (test helper).

set_pur_response ¶

set_pur_response(
    public_identity: str, result_code: int = 2001
) -> None

Configure a mock PUA response for a specific user (test helper).

set_snr_response ¶

set_snr_response(
    public_identity: str, result_code: int = 2001
) -> None

Configure a mock SNA response for a specific user (test helper).

clear ¶

clear() -> None

Reset all mock peers and responses (test helper).

`DiameterRequest`¶

The inbound request passed to @diameter.on_request.

Mock DiameterRequest passed to @diameter.on_request in tests.

Construct one in your test and invoke your handler with it.

answer ¶

answer(
    result_code: int = 2001,
    error_message: Optional[str] = None,
) -> MockDiameterAnswer

Build a local answer to serve this request (HSS-style). Populate it with :meth:MockDiameterAnswer.set_avp, including grouped AVPs (pass a list of (code, value[, vendor]) child tuples as the value).

forward_to `async` ¶

forward_to(
    peer: MockPeer,
    identity=None,
    timeout_secs: float = 10.0,
) -> MockDiameterAnswer

Mock relay — returns a 2001 success answer (override in tests by monkeypatching if a different result is needed).

`DiameterAnswer`¶

The value a handler returns via request.answer(...) / request.reject(...).

Mock DiameterAnswer — the value a handler returns / forwards.

Diameter¶

diameter namespace¶

config property ¶

event_sink property ¶

is_connected ¶

peer_count ¶

decode_isdn_address ¶

encode_isdn_address ¶

cx_uar ¶

cx_sar ¶

cx_lir ¶

rx_aar ¶

rx_str ¶

sh_udr ¶

sh_pur ¶

sh_snr ¶

add_peer ¶

set_default_server_name ¶

set_uar_response ¶

set_sar_response ¶

set_lir_response ¶

set_aar_response ¶

rf_acr_start ¶

rf_acr_interim ¶

rf_acr_stop ¶

rf_acr_event ¶

set_rf_result_code ¶

set_rf_interim_interval ¶

captured_acrs ¶

clear_captured_acrs ¶

ro_ccr_initial async ¶

ro_ccr_update async ¶

ro_ccr_terminate async ¶

ro_ccr_event async ¶

set_ro_result_code ¶

set_ro_granted_time ¶

set_ro_final_unit_action ¶

captured_ccrs ¶

clear_captured_ccrs ¶

on_inbound_cer staticmethod ¶

on_request staticmethod ¶

on_reply staticmethod ¶

on_request_completed staticmethod ¶

peer_pool ¶

ip_in_cidr staticmethod ¶

fnmatch staticmethod ¶

now_us staticmethod ¶

set_config ¶

s6a_air ¶

s6a_ulr ¶

s6a_purge_ue ¶

s6c_srr ¶

s6c_rsr ¶

sgd_tfr ¶

send_request ¶

set_generic_response ¶

set_udr_response ¶

set_pur_response ¶

set_snr_response ¶

clear ¶

DiameterRequest¶

answer ¶

forward_to async ¶

DiameterAnswer¶

`diameter` namespace¶

config `property` ¶

event_sink `property` ¶

ro_ccr_initial `async` ¶

ro_ccr_update `async` ¶

ro_ccr_terminate `async` ¶

ro_ccr_event `async` ¶

on_inbound_cer `staticmethod` ¶

on_request `staticmethod` ¶

on_reply `staticmethod` ¶

on_request_completed `staticmethod` ¶

ip_in_cidr `staticmethod` ¶

fnmatch `staticmethod` ¶

now_us `staticmethod` ¶

`DiameterRequest`¶

forward_to `async` ¶

`DiameterAnswer`¶