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

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.

reset ¶

reset() -> None

Reset all mock state between tests.

Clears: handlers, registrar, auth, cache, log, rtpengine.

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.

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.

send_answer ¶

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

Send a B2BUA answer event.

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.

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.

`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.

Testing harness¶

SipTestHarness¶

registrar property ¶

auth property ¶

log property ¶

cache property ¶

rtpengine property ¶

proxy property ¶

b2bua property ¶

presence property ¶

diameter property ¶

gateway property ¶

li property ¶

hss property ¶

pcrf property ¶

reset ¶

load_script ¶

load_source ¶

send_request ¶

send_reply ¶

send_cancel ¶

send_invite ¶

send_answer ¶

send_failure ¶

send_bye ¶

send_call_cancel ¶

send_refer ¶

close ¶

RequestResult¶

request instance-attribute ¶

actions class-attribute instance-attribute ¶

action property ¶

status_code property ¶

reason property ¶

targets property ¶

strategy property ¶

next_hop property ¶

was_relayed property ¶

was_forked property ¶

was_dropped property ¶

record_routed property ¶

ReplyResult¶

reply instance-attribute ¶

CallResult¶

call instance-attribute ¶

MockHss¶

add_subscriber ¶

remove_subscriber ¶

set_auth_failure ¶

subscriber_count ¶

clear ¶

MockPcrf¶

accept_all ¶

reject_all ¶

reject_session ¶

clear ¶

`SipTestHarness`¶

registrar `property` ¶

auth `property` ¶

log `property` ¶

cache `property` ¶

rtpengine `property` ¶

proxy `property` ¶

b2bua `property` ¶

presence `property` ¶

diameter `property` ¶

gateway `property` ¶

li `property` ¶

hss `property` ¶

pcrf `property` ¶

`RequestResult`¶

request `instance-attribute` ¶

actions `class-attribute` `instance-attribute` ¶

action `property` ¶

status_code `property` ¶

reason `property` ¶

targets `property` ¶

strategy `property` ¶

next_hop `property` ¶

was_relayed `property` ¶

was_forked `property` ¶

was_dropped `property` ¶

record_routed `property` ¶

`ReplyResult`¶

reply `instance-attribute` ¶

`CallResult`¶

call `instance-attribute` ¶

`MockHss`¶

`MockPcrf`¶