siphon-sip

Mock library and type stubs for SIPhon scripts — enables unit testing without the Rust binary and provides rich context for LLM-assisted script authoring.

Install

pip install siphon-sip

The PyPI distribution is siphon-sip (matching the siphon-sip crate). The import package is still siphon_sdk — from siphon_sdk import ….

What is SIPhon?

SIPhon is a high-performance SIP proxy, B2BUA, and IMS platform written in Rust with Python scripting. Scripts use decorators to handle SIP events:

from siphon import proxy, registrar, auth, log

@proxy.on_request
def route(request):
    if request.method == "REGISTER":
        if not auth.require_digest(request, realm="example.com"):
            return
        registrar.save(request)
        request.reply(200, "OK")
        return

    contacts = registrar.lookup(request.ruri)
    if not contacts:
        request.reply(404, "Not Found")
        return

    request.record_route()
    request.fork([c.uri for c in contacts])

This SDK lets you test these scripts with pytest — no Rust binary needed.

Quick start

from siphon_sdk.testing import SipTestHarness
from siphon_sdk.types import Contact

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

# Pre-populate the registrar
harness.registrar.add_contact(
    "sip:alice@example.com",
    Contact(uri="sip:alice@192.168.1.5:5060"),
)

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

# Test INVITE routing
result = harness.send_request("INVITE", "sip:alice@example.com")
assert result.action == "fork"
assert "sip:alice@192.168.1.5:5060" in result.targets

Testing B2BUA scripts

harness = SipTestHarness()
harness.load_script("scripts/b2bua_default.py")

harness.registrar.add_contact(
    "sip:bob@example.com",
    Contact(uri="sip:bob@10.0.0.2:5060"),
)

result = harness.send_invite(ruri="sip:bob@example.com")
assert result.action == "fork"
assert result.targets == ["sip:bob@10.0.0.2:5060"]

# Test BYE handling
result = harness.send_bye(initiator_side="a")
assert result.was_terminated

Testing extension scripts (SMPP, HTTP)

The opt-in siphon extensions inject extra namespaces (smpp, http) at runtime. The SDK mocks them too, with dedicated harnesses, so extension scripts are testable from the same pip install siphon-sip — no running SMSC or HTTP listener required:

from siphon_sdk.smpp_testing import SmppTestHarness

harness = SmppTestHarness()
harness.load_script("scripts/gateway.py")
assert harness.bind("esme1", password="s3cret")
reply = harness.submit_sm(source_addr="15550100", destination_addr="15550101",
                          short_message=b"hi")
assert reply.ok

from siphon_sdk.http_testing import HttpTestHarness
from siphon_sdk.http import MockResponse

harness = HttpTestHarness()
harness.add_response(MockResponse(status=200, body=b'{"ok":true}'))
harness.load_script("scripts/api.py")

resp = harness.request("GET", "/users/42")
assert resp.status == 200

Inline scripts

Test scripts without separate files:

harness = SipTestHarness()
harness.load_source("""
from siphon import proxy

@proxy.on_request
def route(request):
    if request.source_ip_in(["10.0.0.0/8"]):
        request.relay()
    else:
        request.reply(403, "Forbidden")
""")

result = harness.send_request("INVITE", "sip:bob@host", source_ip="10.1.2.3")
assert result.was_relayed

result = harness.send_request("INVITE", "sip:bob@host", source_ip="8.8.8.8")
assert result.status_code == 403

Async handlers + RTPEngine

harness = SipTestHarness()
harness.load_source("""
from siphon import proxy, rtpengine

@proxy.on_request
async def route(request):
    if request.method == "INVITE" and request.body:
        await rtpengine.offer(request, profile="srtp_to_rtp")
    request.relay()
""")

result = harness.send_request("INVITE", "sip:bob@host",
                              body=b"v=0\\r\\n...",
                              content_type="application/sdp")
assert result.was_relayed
assert harness.rtpengine.operations == [("offer", "srtp_to_rtp")]

Controlling mock behavior

# Auth: allow or deny all
harness.auth._allow = True  # all auth checks pass

# Rate limiting
harness.proxy._utils._rate_limit_allow = False  # simulate overload

# Cache: pre-populate
harness.cache.set_data("cnam", {"key": "value"})

# Registrar: add contacts directly
harness.registrar.add_contact("sip:alice@host", Contact(uri="sip:alice@1.2.3.4"))

# Log: inspect captured messages
assert any("error" in msg for level, msg in harness.log.messages)

# Reset between tests
harness.reset()

Result assertions

RequestResult provides convenient properties:

API reference

Namespaces

Request properties

Request methods

Registrar

lookup() returns Contact objects:

Auth

B2BUA call

Each B-leg gets a fresh Call-ID and From-tag by default, fully decoupling the two SIP dialogs. Use keep_call_id() to opt out of Call-ID regeneration.

License

MIT