Testing harness

The siphon_sdk.testing module lets you unit-test SIPhon scripts with pytest, no Rust binary required. The harness installs the mock siphon module, loads your script, feeds it simulated SIP messages, and returns a result object that records what the handler did.

from siphon_sdk.testing import SipTestHarness

def test_register_challenges_unauthenticated():
    harness = SipTestHarness()
    harness.load_script("scripts/proxy_default.py")

    result = harness.send_request("REGISTER", "sip:alice@example.com")
    assert result.action == "reply"
    assert result.status_code == 401

SipTestHarness

High-level test harness for SIPhon scripts.

Usage::

harness = SipTestHarness(local_domains=["example.com"])
harness.load_script("scripts/proxy_default.py")

result = harness.send_request("REGISTER", "sip:alice@example.com",
                              from_uri="sip:alice@example.com")
assert result.status_code == 401

The harness: 1. Installs the mock siphon module 2. Loads and executes user scripts (registering their decorators) 3. Dispatches mock SIP messages to registered handlers 4. Returns structured results for assertion

inbound_calls_active property

inbound_calls_active: int

Calls currently holding a slot — what siphon reports as siphon_b2bua_calls_active.

registrar property

registrar: MockRegistrar

Access the mock registrar to pre-populate contacts.

auth property

auth: MockAuth

Access the mock auth to control authentication behavior.

log property

log: MockLog

Access captured log messages.

cache property

cache: MockCache

Access the mock cache to pre-populate data.

rtpengine property

rtpengine: MockRtpEngine

Access the mock RTPEngine to inspect operations.

proxy property

proxy: MockProxy

Access the mock proxy namespace (e.g. for _utils config).

b2bua property

b2bua: MockB2bua

Access the mock b2bua namespace to inspect terminates.

presence property

presence: MockPresence

Access the mock presence store.

diameter property

diameter: MockDiameter

Access the mock Diameter namespace (Cx/Rx operations).

gateway property

gateway: MockGateway

Access the mock gateway namespace.

li property

li: MockLi

Access the mock lawful intercept namespace.

hss property

hss: MockHss

Access a convenience MockHss wired to this harness.

Lazily created on first access.

pcrf property

pcrf: MockPcrf

Access a convenience MockPcrf wired to this harness.

Lazily created on first access.

set_inbound_limit

set_inbound_limit(
    max_concurrent_calls: Optional[int] = None,
    max_calls_per_second: Optional[int] = None,
    reject_code: int = 503,
    retry_after_secs: int = 1,
) -> None

Enforce b2bua.inbound_limit on :meth:send_invite.

Mirrors what siphon does with::

b2bua:
  inbound_limit:
    max_concurrent_calls: 2
    max_calls_per_second: 10

A call past either ceiling is refused before @b2bua.on_invite runs: no handler fires for it, so a script cannot observe, log or re-route a refused call. The result has :attr:CallResult.was_refused set and the refusal's status in :attr:CallResult.status_code.

Two things to size the ceilings by:

• max_calls_per_second counts INVITEs. A call the script challenges for digest credentials arrives twice (the 407, then the retry), so it spends two.
• The ceilings are per siphon instance.

A slot is released when the call ends: :meth:send_bye, :meth:send_call_cancel, a :meth:send_failure the handler does not route again, or an on_invite that rejects, terminates or does nothing. An emergency call (request URI urn:service:sos, RFC 5031) is counted and never refused.

The rate runs on the harness's own clock, which only :meth:advance_time moves, so a test is deterministic::

harness.set_inbound_limit(max_calls_per_second=1)
assert not harness.send_invite().was_refused
assert harness.send_invite().was_refused
harness.advance_time(1.0)
assert not harness.send_invite().was_refused

Parameters:

Name	Type	Description	Default
max_concurrent_calls	Optional[int]	Calls held at once. None or 0 is unlimited.	None
max_calls_per_second	Optional[int]	New calls per second, with a burst of one second's worth. None or 0 is unlimited.	None
reject_code	int	Status a refused call is answered with (400-699).	503
retry_after_secs	int	Retry-After on the refusal; 0 omits it.	1

Raises:

Type	Description
ValueError	reject_code is not a failure response.

clear_inbound_limit

clear_inbound_limit() -> None

Stop enforcing an inbound limit. Calls in progress stay counted.

advance_time

advance_time(seconds: float) -> None

Move the clock every max_calls_per_second is measured against, this limit's and each gateway group's.

reset

reset() -> None

Reset all mock state between tests.

Clears: handlers, registrar, auth, cache, log, rtpengine, and the inbound limit with every slot held against it.

load_script

load_script(path: str) -> None

Load and execute a SIPhon script, registering its handlers.

Parameters:

Name	Type	Description	Default
path	str	Path to the Python script file.	required

Raises:

Type	Description
FileNotFoundError	If the script file doesn't exist.

load_source

load_source(source: str, name: str = '<test>') -> None

Load a script from a string (useful for inline test scripts).

Parameters:

Name	Type	Description	Default
source	str	Python source code.	required
name	str	Module name for error messages.	'<test>'

send_request

send_request(
    method: str = "INVITE",
    ruri: Union[str, SipUri] = "sip:bob@example.com",
    *,
    from_uri: Union[
        str, SipUri, None
    ] = "sip:alice@example.com",
    to_uri: Union[str, SipUri, None] = None,
    from_tag: Optional[str] = None,
    to_tag: Optional[str] = None,
    call_id: Optional[str] = None,
    cseq: Optional[tuple[int, str]] = None,
    max_forwards: int = 70,
    body: Optional[bytes] = None,
    content_type: Optional[str] = None,
    transport: str = "udp",
    source_ip: str = "127.0.0.1",
    user_agent: Optional[str] = None,
    auth_user: Optional[str] = None,
    contact_expires: Optional[int] = None,
    event: Optional[str] = None,
    headers: Optional[dict[str, str]] = None
) -> RequestResult

Send a mock SIP request and return the handler's result.

Dispatches to handlers registered via @proxy.on_request.

Parameters:

Name	Type	Description	Default
method	str	SIP method (e.g. "INVITE", "REGISTER").	'INVITE'
ruri	Union[str, SipUri]	Request-URI.	'sip:bob@example.com'
from_uri	Union[str, SipUri, None]	From header URI.	'sip:alice@example.com'
to_uri	Union[str, SipUri, None]	To header URI (defaults to ruri).	None
from_tag	Optional[str]	From-tag (auto-generated if None).	None
to_tag	Optional[str]	To-tag (None for initial requests).	None
call_id	Optional[str]	Call-ID (auto-generated if None).	None
cseq	Optional[tuple[int, str]]	CSeq tuple (auto-generated if None).	None
max_forwards	int	Max-Forwards value.	70
body	Optional[bytes]	Message body bytes.	None
content_type	Optional[str]	Content-Type header.	None
transport	str	Transport protocol.	'udp'
source_ip	str	Source IP address.	'127.0.0.1'
user_agent	Optional[str]	User-Agent header.	None
auth_user	Optional[str]	Pre-authenticated username.	None
contact_expires	Optional[int]	Contact expires value.	None
event	Optional[str]	Event header value.	None
headers	Optional[dict[str, str]]	Additional headers dict.	None

Returns:

Type	Description
RequestResult	class:RequestResult with action details and assertions.

send_reply

send_reply(
    request: Optional[Request] = None,
    status_code: int = 200,
    reason: str = "OK",
    *,
    from_uri: Union[str, SipUri, None] = None,
    to_uri: Union[str, SipUri, None] = None,
    call_id: Optional[str] = None,
    body: Optional[bytes] = None,
    content_type: Optional[str] = None,
    headers: Optional[dict[str, str]] = None
) -> ReplyResult

Send a mock SIP reply and return the handler's result.

Dispatches to handlers registered via @proxy.on_reply whose method filter matches request.method (unfiltered handlers always run).

Parameters:

Name	Type	Description	Default
request	Optional[Request]	Original request (auto-created if None).	None
status_code	int	SIP status code (e.g. 200, 404).	200
reason	str	Reason phrase.	'OK'
from_uri	Union[str, SipUri, None]	From URI.	None
to_uri	Union[str, SipUri, None]	To URI.	None
call_id	Optional[str]	Call-ID.	None
body	Optional[bytes]	Response body.	None
content_type	Optional[str]	Content-Type.	None
headers	Optional[dict[str, str]]	Additional headers.	None

Returns:

Type	Description
ReplyResult	class:ReplyResult with action details.

send_cancel

send_cancel(
    request: Optional[Request] = None, **kwargs: Any
) -> RequestResult

Dispatch a CANCEL teardown to @proxy.on_cancel handlers.

Simulates a relayed INVITE being CANCELled before any final response. The handler receives the original INVITE request. Fire-and-forget: there is no reply/relay gating — assert on side effects (e.g. harness.rtpengine deletes, harness.diameter Rx STR).

Parameters:

Name	Type	Description	Default
request	Optional[Request]	Original INVITE (auto-created with method="INVITE" if None).	None
**kwargs	Any	Passed to the :class:Request constructor when auto-creating.	{}

send_invite

send_invite(
    call: Optional[Call] = None, **kwargs: Any
) -> CallResult

Send a B2BUA INVITE event.

Parameters:

Name	Type	Description	Default
call	Optional[Call]	Call object (auto-created if None).	None
**kwargs	Any	Passed to :class:Call constructor.	{}

Returns:

Type	Description
CallResult	class:CallResult with action details. With an inbound limit set
CallResult	(:meth:set_inbound_limit) and reached, the call is refused and no
CallResult	handler runs — see :attr:CallResult.was_refused.

dispatch_invite

dispatch_invite(
    ruri: Union[str, SipUri] = "sip:bob@example.com",
    *,
    call_id: Optional[str] = None,
    from_uri: Union[
        str, SipUri, None
    ] = "sip:alice@example.com",
    to_uri: Union[str, SipUri, None] = None,
    to_tag: Optional[str] = None,
    body: Optional[bytes] = None,
    headers: Optional[dict[str, str]] = None,
    source_ip: str = "127.0.0.1"
) -> Union[CallResult, RequestResult]

Send an INVITE down the path siphon would take it.

A script whose @b2bua.on_invite handlers all carry a request_uri= filter is a B2BUA for the INVITEs a filter matches and a proxy for the rest. This sends an initial INVITE (no to_tag) the B2BUA takes to :meth:send_invite and returns its :class:CallResult; any other INVITE, an in-dialog one included, goes to :meth:send_request and returns its :class:RequestResult. With unfiltered B2BUA handlers every INVITE is the B2BUA's, and with none every INVITE is the proxy's, as at runtime.

Example::

result = harness.dispatch_invite("sip:transfer@example.com")
assert isinstance(result, CallResult)

send_answer

send_answer(
    call: Optional[Call] = None, **kwargs: Any
) -> CallResult

Send a B2BUA answer event.

send_provisional

send_provisional(
    call: Optional[Call] = None,
    reply: Optional[Any] = None,
    status_code: int = 180,
    reason: str = "Ringing",
    **kwargs: Any
) -> CallResult

Send a provisional response of the callee (101-199) on a B2BUA call.

Runs @b2bua.on_provisional handlers, then @b2bua.on_early_media handlers when the reply has a body, on the same reply: what siphon does for a provisional it relays to the caller. A handler may call reply.relay(body=False) or reply.withhold() on it; reply.provisional_relay reads the decision back.

Parameters:

Name	Type	Description	Default
call	Optional[Call]	Call object (auto-created from **kwargs if None).	None
reply	Optional[Any]	The callee's response. Pass one to read back what the handlers set on it; auto-created from status_code and reason if None.	None
status_code	int	Status of the auto-created reply.	180
reason	str	Reason phrase of the auto-created reply.	'Ringing'

send_failure

send_failure(
    call: Optional[Call] = None,
    code: int = 486,
    reason: str = "Busy Here",
    **kwargs: Any
) -> CallResult

Send a B2BUA failure event.

Parameters:

Name	Type	Description	Default
call	Optional[Call]	Call object.	None
code	int	Failure status code.	486
reason	str	Failure reason phrase.	'Busy Here'

send_bye

send_bye(
    call: Optional[Call] = None,
    initiator_side: str = "a",
    **kwargs: Any
) -> CallResult

Send a B2BUA BYE event.

Parameters:

Name	Type	Description	Default
call	Optional[Call]	Call object.	None
initiator_side	str	"a" (caller) or "b" (callee).	'a'

send_call_cancel

send_call_cancel(
    call: Optional[Call] = None, **kwargs: Any
) -> CallResult

Dispatch a CANCEL teardown to @b2bua.on_cancel handlers.

Simulates an unanswered call (Calling/Ringing) being CANCELled. The handler receives the :class:Call. Fire-and-forget — assert on side effects (e.g. harness.rtpengine deletes).

Parameters:

Name	Type	Description	Default
call	Optional[Call]	Call object (auto-created in state="ringing" if None).	None
**kwargs	Any	Passed to the :class:Call constructor when auto-creating.	{}

send_refer

send_refer(
    call: Optional[Call] = None,
    refer_to: str = "sip:+15550111@example.com",
    refer_replaces: Optional[dict] = None,
    **kwargs: Any
) -> CallResult

Dispatch a REFER (call transfer) to @b2bua.on_refer handlers.

Mirrors :meth:send_invite for the transfer path. Builds an answered :class:Call carrying the transfer target on :attr:Call.refer_to (and :attr:Call.refer_replaces for an attended transfer), then dispatches it to the registered @b2bua.on_refer handler with a single (call,) argument — a REFER is a request, so there is no reply object. Supports both sync and async handlers.

Note

With no registered handler this returns an empty :class:CallResult (no action recorded). In production the Rust dispatcher locally rejects an un-handled REFER (501 / 603) — that default is a dispatcher concern, out of scope for the mock.

Parameters:

Name	Type	Description	Default
call	Optional[Call]	Call object (auto-created answered if None).	None
refer_to	str	The Refer-To URI (transfer target).	'sip:+15550111@example.com'
refer_replaces	Optional[dict]	Optional attended-transfer dict with call_id / from_tag / to_tag (and optional early_only); None for a blind transfer.	None
**kwargs	Any	Passed to the :class:Call constructor when auto-creating.	{}

Returns:

Type	Description
CallResult	class:CallResult with the actions the handler took.

send_reoffer

send_reoffer(
    call: Optional[Call] = None,
    sdp: Optional[str] = None,
    previous_sdp: Optional[str] = None,
    method: str = "INVITE",
    side: str = "a",
    early: bool = False,
    request: Optional[Request] = None,
    session_sdp: Optional[str] = None,
    **kwargs: Any
) -> Reoffer

Dispatch an in-dialog offer to @b2bua.on_reoffer handlers.

Builds a :class:~siphon_sdk.reoffer.Reoffer for a re-INVITE (or an UPDATE with method="UPDATE") from side carrying sdp, read against previous_sdp (the session that party had in force), with session_sdp as what siphon last sent that party, and calls the handlers with (call, offer). Returns the offer: its :attr:~siphon_sdk.reoffer.Reoffer.decision is ("relay",), ("reject", code, reason) or ("answer", sdp). A handler that raises leaves the decision at relay, as siphon does.

Example::

offer = harness.send_reoffer(sdp=HELD, previous_sdp=TALKING)
assert offer.kind == "hold"
assert offer.decision == ("answer", None)

send_reoffer_answered

send_reoffer_answered(
    offer: Reoffer,
    status_code: int = 200,
    answer_sdp: Optional[str] = None,
    answered_locally: bool = False,
    call: Optional[Call] = None,
    **kwargs: Any
) -> Reoffer

Dispatch the final response to offer to @b2bua.on_reoffer_answered handlers.

A raise in a handler is swallowed, as siphon logs it and goes on.

close

close() -> None

Clean up the event loop.

RequestResult

The result of harness.send_request(...).

Result of sending a request through the harness.

Provides convenient access to the primary action the handler took, plus the full list of actions and the request/reply objects.

request instance-attribute

request: Request

The request object that was passed to the handler.

actions class-attribute instance-attribute

actions: list[Action] = field(default_factory=list)

All actions the handler took (reply, relay, fork, etc.).

action property

action: str

The primary action kind (last action), or "silent_drop".

status_code property

status_code: Optional[int]

Status code from the last reply/reject action, or None.

reason property

reason: Optional[str]

Reason phrase from the last reply/reject action.

targets property

targets: Optional[list[str]]

Fork/dial targets from the last fork action.

strategy property

strategy: Optional[str]

Fork strategy from the last fork action.

next_hop property

next_hop: Optional[str]

Next-hop from the last relay action.

was_relayed property

was_relayed: bool

True if the handler called relay().

was_forked property

was_forked: bool

True if the handler called fork().

was_dropped property

was_dropped: bool

True if the handler returned without any terminal action (silent drop semantics).

record_routed property

record_routed: bool

True if record_route() was called.

ReplyResult

The result of feeding a response through @proxy.on_reply.

Result of sending a reply through the harness.

reply instance-attribute

reply: Reply

The reply object passed to the handler.

CallResult

The result of driving a B2BUA call.

Result of sending a B2BUA event through the harness.

call instance-attribute

call: Call

The call object passed to the handler.

was_refused property

was_refused: bool

True when siphon refused the call at admission (b2bua.inbound_limit), before @b2bua.on_invite ran.

The refusal's status is :attr:status_code. No handler saw the call, so there is nothing else on the result::

harness.set_inbound_limit(max_concurrent_calls=1)
harness.send_invite()
result = harness.send_invite()
assert result.was_refused and result.status_code == 503

retry_after_secs class-attribute instance-attribute

retry_after_secs: Optional[int] = None

On a refused call, the Retry-After siphon sent, in seconds. None when the call was not refused or the header was turned off.

refusal_scope class-attribute instance-attribute

refusal_scope: Optional[str] = None

On a refused call, which limit refused it: "global" for b2bua.inbound_limit, "gateway" for a gateway group's own.

refusal_gateway_group class-attribute instance-attribute

refusal_gateway_group: Optional[str] = None

On a call a gateway group's limit refused, that group's name.

MockHss

An in-process HSS stub for exercising Diameter Cx/Sh flows.

High-level mock HSS for IMS script testing.

Combines subscriber data management with automated Diameter Cx response generation. Wires into MockDiameter and MockAuth so test scripts get realistic IMS registration and routing flows.

Usage::

harness = SipTestHarness(local_domains=["ims.example.com"])
hss = harness.hss

hss.add_subscriber(
    impi="alice@ims.example.com",
    impu="sip:alice@ims.example.com",
    server_name="sip:scscf.ims.example.com:6060",
)
harness.load_script("examples/ims_icscf.py")

result = harness.send_request("REGISTER", "sip:ims.example.com",
                               from_uri="sip:alice@ims.example.com")
assert result.was_relayed
assert result.next_hop == "sip:scscf.ims.example.com:6060"

add_subscriber

add_subscriber(
    impi: str,
    impu: str,
    server_name: Optional[str] = None,
    ifc_xml: Optional[str] = None,
) -> None

Register a subscriber in the mock HSS.

This configures the mock Diameter Cx responses so that: - diameter.cx_uar(impu) returns the assigned server_name - diameter.cx_lir(impu) returns the serving server_name - diameter.cx_sar(impu) returns success with ifc_xml - auth.require_aka_digest() / auth.require_ims_digest() auto-passes (returns True)

Parameters:

Name	Type	Description	Default
impi	str	IMS Private Identity (e.g. "alice@ims.example.com").	required
impu	str	IMS Public Identity (e.g. "sip:alice@ims.example.com").	required
server_name	Optional[str]	Assigned S-CSCF URI.	None
ifc_xml	Optional[str]	Initial Filter Criteria XML for the user profile.	None

remove_subscriber

remove_subscriber(impu: str) -> None

Remove a subscriber from the mock HSS.

Parameters:

Name	Type	Description	Default
impu	str	IMS Public Identity to remove.	required

set_auth_failure

set_auth_failure(impu: str) -> None

Configure authentication to fail for a subscriber.

Parameters:

Name	Type	Description	Default
impu	str	IMS Public Identity.	required

subscriber_count

subscriber_count() -> int

Number of registered subscribers.

clear

clear() -> None

Remove all subscribers and reset Diameter/auth state.

MockPcrf

An in-process PCRF stub for exercising Diameter Rx flows.

High-level mock PCRF for IMS P-CSCF testing.

Simulates a Policy and Charging Rules Function for Rx interface testing. Configures MockDiameter Rx responses for QoS resource operations.

Usage::

harness = SipTestHarness(local_domains=["ims.example.com"])
pcrf = harness.pcrf

# PCRF will accept all AAR requests
pcrf.accept_all()

# Or reject specific sessions
pcrf.reject_session("rx-sess-1", result_code=5003)

accept_all

accept_all() -> None

Configure the PCRF to accept all Rx AAR requests with 2001.

reject_all

reject_all(result_code: int = 5003) -> None

Configure the PCRF to reject all Rx AAR requests.

Parameters:

Name	Type	Description	Default
result_code	int	Diameter result code for rejection (default 5003 = DIAMETER_AUTHORIZATION_REJECTED).	5003

reject_session

reject_session(
    session_id: str, result_code: int = 5003
) -> None

Reject a specific Rx session.

Parameters:

Name	Type	Description	Default
session_id	str	Rx session ID to reject.	required
result_code	int	Diameter result code.	5003

clear

clear() -> None

Reset PCRF mock state.