Observability¶

Cross-cutting utility namespaces: structured logging (log), the named cache backends (cache), call detail records (cdr), custom Prometheus metrics (metrics), and periodic / one-shot timers (timer).

`log` namespace¶

Mock logging namespace — captures log messages for test assertions.

Access captured messages via log.messages::

from siphon import log
log.info("hello")
assert ("info", "hello") in log.messages

messages `instance-attribute` ¶

messages: list[tuple[str, str]] = []

List of (level, message) tuples captured during the test.

debug ¶

debug(msg: str) -> None

Log at DEBUG level.

info ¶

info(msg: str) -> None

Log at INFO level.

warn ¶

warn(msg: str) -> None

Log at WARN level.

warning ¶

warning(msg: str) -> None

Alias for :meth:warn.

error ¶

error(msg: str) -> None

Log at ERROR level.

clear ¶

clear() -> None

Clear all captured messages (test helper).

`cache` namespace¶

Named cache backends (Redis + local LRU) from the cache: config list.

Mock cache namespace with an in-memory dict backend.

Pre-populate::

from siphon import cache
cache.set_data("cnam", {"msisdn_display:1234": "Sales"})

Then await cache.fetch("cnam", "msisdn_display:1234") returns "Sales".

fetch `async` ¶

fetch(name: str, key: str) -> Optional[str]

Fetch a value from a named cache.

Parameters:

Name	Type	Description	Default
`name`	`str`	Cache name (from `siphon.yaml` `cache:` list).	required
`key`	`str`	Cache key string.	required

Returns:

Type	Description
`Optional[str]`	Cached value or `None` if not found.

store `async` ¶

store(
    name: str,
    key: str,
    value: str,
    ttl: Optional[int] = None,
) -> bool

Store a value in a named cache with optional TTL.

Parameters:

Name	Type	Description	Default
`name`	`str`	Cache name.	required
`key`	`str`	Cache key.	required
`value`	`str`	Value to store.	required
`ttl`	`Optional[int]`	Optional TTL in seconds. Mirrors the real `cache.store` signature; the mock is in-memory and does not expire keys, so the value is accepted and ignored.	`None`

Returns:

Type	Description
`bool`	`True` if stored, `False` if cache name unknown.

has_cache ¶

has_cache(name: str) -> bool

Check if a named cache exists.

delete `async` ¶

delete(name: str, key: str) -> bool

Delete a key. Returns True if the cache exists.

exists `async` ¶

exists(name: str, key: str) -> bool

Check if key is present in the named cache.

list_push `async` ¶

list_push(name: str, key: str, item: str) -> Optional[int]

Append item to a list under key. Returns the new length, or None if the cache name is unknown.

The mock stores lists as Python list[str] under the same keyspace as scalars — fine for tests, but a real script would not mix scalar and list values on the same key.

list_pop_all `async` ¶

list_pop_all(name: str, key: str) -> list[str]

Atomically read and clear a list. Returns the items in FIFO order, empty list when the key was absent or the cache is unknown.

list_len `async` ¶

list_len(name: str, key: str) -> Optional[int]

Return the length of the list under key (0 for a missing key), or None if the cache name is unknown.

list_len_sum `async` ¶

list_len_sum(name: str, prefix: str) -> Optional[int]

Sum the lengths of every list whose key starts with prefix. Returns 0 when nothing matches, None if the cache name is unknown. Raises ValueError on an empty prefix (which would scan the entire keyspace).

expire `async` ¶

expire(name: str, key: str, ttl: int) -> bool

Mock TTL — records the call on self.expirations for assertions and returns True when the key currently exists in the cache. The mock does not actually time out entries.

set_data ¶

set_data(
    name: str, data: Optional[dict[str, str]] = None
) -> None

Create/replace a named cache with test data (test helper).

Parameters:

Name	Type	Description	Default
`name`	`str`	Cache name.	required
`data`	`Optional[dict[str, str]]`	Initial key-value pairs (default: empty dict).	`None`

clear ¶

clear() -> None

Remove all caches (test helper).

`cdr` namespace¶

Call detail record writing from scripts.

Mock cdr namespace — call detail record writing from scripts.

Usage::

from siphon import cdr

cdr.write(request, extra={"billing_id": "B-12345"})  # proxy handler
cdr.write(call, extra={"billing_id": "B-12345"})     # b2bua handler
cdr.enabled  # True if CDR system is active

Test helper::

from siphon_sdk.mock_module import get_cdr
cdrs = get_cdr().records  # list of written CDR dicts

With cdr.auto_emit on, the engine does not write a second record: it merges extra into the record it is already tracking for the call, so the fields are emitted once at teardown alongside the timings and the disconnect side. Set auto_emit = True on the mock to test a script against that shape::

cdr_mock = get_cdr()
cdr_mock.auto_emit = True
...                                     # run the handler
assert cdr_mock.pending[call.id]["billing_id"] == "B-12345"
cdr_mock.finalize(call)                 # stands in for the BYE
assert cdr_mock.records[0]["billing_id"] == "B-12345"

enabled `property` ¶

enabled: bool

Whether the CDR system is enabled.

typed_records `property` ¶

typed_records: list[CallDetailRecord]

The written records as :class:siphon_sdk.cdr.CallDetailRecord — the same type a collector receives from the HTTP/file sinks.

Assert against this rather than raw dicts when the test cares about the shape the outside world sees::

record = get_cdr().typed_records[-1]
assert record.extra["billing_id"] == "B-12345"

session_key `staticmethod` ¶

session_key(source: 'Any') -> str

The key the engine tracks this call's auto-emit record under.

A B2BUA call is keyed on its internal call id (one record for two dialogs); a proxy call on <Call-ID>\0<From-tag>.

finalize ¶

finalize(source: 'Any', **fields: 'Any') -> 'dict | None'

Emit the record tracked for source — the mock's stand-in for the BYE that ends the call. Returns the emitted record, or None when nothing was tracked.

write ¶

write(
    source: "Any", extra: "dict[str, str] | None" = None
) -> bool

Write a CDR for the given request or B2BUA call — or attach the fields to the record siphon is already keeping for it.

With cdr.auto_emit on (auto_emit = True on this mock), extra is merged into the tracked record instead of queueing a second one, so the fields are emitted once at teardown on the record that also carries the timings, duration_secs, response_code and disconnect_initiator. A standalone record is still written when there is nothing to merge into.

Parameters:

Name	Type	Description	Default
`source`	`'Any'`	The SIP `Request` (proxy handlers) OR the B2BUA `Call` (`@b2bua.on_answer` / `on_bye` / … handlers). Both carry the Call-ID, From/To/R-URI and source IP the CDR needs.	required
`extra`	`'dict[str, str] \| None'`	Optional dict of extra fields to include in the CDR.	`None`

Returns:

Type	Description
`bool`	True if the fields reached a CDR — merged into the tracked record,
`bool`	or queued as a record of their own.

Raises:

Type	Description
`TypeError`	if `source` is neither a `Request` nor a `Call`.

Example::

from siphon import cdr

@proxy.on_request("INVITE")
def route(request):
    cdr.write(request, extra={"billing_id": "B-12345"})

@b2bua.on_answer
def answered(call, reply):
    cdr.write(call, extra={"billing_id": "B-12345"})

clear ¶

clear() -> None

Reset CDR records (test helper).

The record a collector receives¶

This namespace is the writing side. What comes out the other end — the JSON object every sink writes, its three record kinds, and the typed siphon_sdk.cdr.CallDetailRecord a collector parses it with — is CDR records.

Writing to several sinks¶

cdr.backends takes a list, and every record is written to every entry:

cdr:
  enabled: true
  auto_emit: true
  backends:
    - type: file
      path: /var/log/siphon/cdr.jsonl
      rotate_size_mb: 100
    - type: http
      url: https://collector.example.com/v1/cdr
      auth_header: "Bearer ..."

This is the usual shape: the file is the durable copy on the node, and the collector is what the billing or reporting side consumes. With one sink, a record the collector fails to take exists nowhere else, and until the collector exists the file is the only place records can go.

Each sink gets its own channel and its own writer task, so a slow or failing sink cannot delay, block or drop another's records. channel_size applies per sink; a full channel drops for that sink alone, logged and counted under its name in siphon_cdr_dropped_total{sink}. Per-sink ordering is preserved.

There is no retry or durable queue for the HTTP sink. A file sink beside it is how a deployment gets durability without one.

The single form — backend: plus its matching file: / syslog: / http: block — is unchanged and means one sink of that kind. Setting both backend and backends is refused at config load rather than guessed at.

`metrics` namespace¶

Custom Prometheus counters, gauges, and histograms that appear on /metrics.

Mock metrics namespace — custom Prometheus metrics from scripts.

Usage::

from siphon import metrics

counter = metrics.counter("bgcf_calls_total", "Total calls",
                          labels=["direction"])
counter.labels(direction="outbound").inc()

gauge = metrics.gauge("bgcf_calls_active", "Active calls")
gauge.inc()

hist = metrics.histogram("bgcf_setup_seconds", "Setup time",
                         buckets=[0.1, 0.5, 1.0])
hist.observe(0.3)

Test helper::

from siphon_sdk.mock_module import get_metrics
m = get_metrics()
c = m.counter("test_total", "Test")
c.inc()
assert c._value == 1.0

counter ¶

counter(
    name: str, help: str, labels: "list[str] | None" = None
) -> MockCounter

Create a new counter metric.

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric name (e.g. `"bgcf_calls_total"`).	required
`help`	`str`	Description string.	required
`labels`	`'list[str] \| None'`	Optional list of label names.	`None`

Returns:

Type	Description
`MockCounter`	A MockCounter handle.

gauge ¶

gauge(
    name: str, help: str, labels: "list[str] | None" = None
) -> MockGauge

Create a new gauge metric.

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric name (e.g. `"bgcf_calls_active"`).	required
`help`	`str`	Description string.	required
`labels`	`'list[str] \| None'`	Optional list of label names.	`None`

Returns:

Type	Description
`MockGauge`	A MockGauge handle.

histogram ¶

histogram(
    name: str,
    help: str,
    labels: "list[str] | None" = None,
    buckets: "list[float] | None" = None,
) -> MockHistogram

Create a new histogram metric.

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric name (e.g. `"bgcf_setup_seconds"`).	required
`help`	`str`	Description string.	required
`labels`	`'list[str] \| None'`	Optional list of label names.	`None`
`buckets`	`'list[float] \| None'`	Optional list of bucket boundaries.	`None`

Returns:

Type	Description
`MockHistogram`	A MockHistogram handle.

clear ¶

clear() -> None

Reset all registered metrics.

`Counter`¶

Mock Prometheus counter.

Usage::

from siphon import metrics

c = metrics.counter("my_total", "My counter")
c.inc()
c.inc(5)

With labels::

c = metrics.counter("my_total", "My counter", labels=["method"])
c.labels(method="INVITE").inc()

inc ¶

inc(n: float = 1.0) -> None

Increment the counter (no-label metrics only).

labels ¶

labels(**kwargs: str) -> _MockMetricChild

Return a labeled child counter.

`Gauge`¶

Mock Prometheus gauge.

Usage::

from siphon import metrics

g = metrics.gauge("my_active", "Active things")
g.inc()
g.dec()
g.set(42)

inc ¶

inc(n: float = 1.0) -> None

Increment the gauge (no-label metrics only).

dec ¶

dec(n: float = 1.0) -> None

Decrement the gauge (no-label metrics only).

set ¶

set(v: float) -> None

Set absolute value (no-label metrics only).

labels ¶

labels(**kwargs: str) -> _MockMetricChild

Return a labeled child gauge.

`Histogram`¶

Mock Prometheus histogram.

Usage::

from siphon import metrics

h = metrics.histogram("my_duration_seconds", "Duration",
                      buckets=[0.1, 0.5, 1.0])
h.observe(0.3)

observe ¶

observe(v: float) -> None

Observe a value (no-label metrics only).

labels ¶

labels(**kwargs: str) -> _MockMetricChild

Return a labeled child histogram.

`timer` namespace¶

Periodic (@timer.every) and one-shot (timer.set) callbacks.

Mock timer namespace for periodic timer callbacks.

Timer handlers run on a Tokio interval in the Rust runtime. They receive no SIP request/call context but can use all other namespaces (registrar, cache, gateway, log, etc.).

Example::

from siphon import timer

@timer.every(seconds=30)
async def health_check():
    for dest in gateway.list("carriers"):
        if not dest.healthy:
            log.warn(f"Gateway {dest.uri} is down")

@timer.every(seconds=300, name="stats_push", jitter=10)
def push_stats():
    log.info("pushing stats")

scheduled `property` ¶

scheduled: dict[str, int]

Map of active one-shot timer keys → scheduled delay (ms).

every ¶

every(
    seconds: int, name: str | None = None, jitter: int = 0
) -> Callable

Register a periodic timer callback.

Parameters:

Name	Type	Description	Default
`seconds`	`int`	Interval between invocations.	required
`name`	`str \| None`	Optional name for logging (defaults to function name).	`None`
`jitter`	`int`	Random jitter in seconds added to each interval (default 0).	`0`

Returns:

Type	Description
`Callable`	Decorator that registers the function as a timer handler.

Example::

@timer.every(seconds=60)
def cleanup():
    presence.expire_stale()

set ¶

set(
    key: str, delay_ms: int, handler: Callable
) -> "MockTimerHandle"

Schedule a one-shot callback under key to fire after delay_ms.

Setting the same key twice cancels the previous timer and reschedules.

In the mock, no tokio runtime fires the callback — tests call :meth:fire with the key to invoke the handler manually.

cancel ¶

cancel(key: str) -> bool

Cancel the one-shot timer registered under key. Returns True if a timer was cancelled, False if no timer matched.

fire ¶

fire(key: str) -> None

Test helper: fire the one-shot timer registered under key.

Raises KeyError if no timer matches. Pops the timer so a subsequent fire for the same key raises.

`TimerHandle`¶

Returned by timer.set(...) — cancel a scheduled one-shot.

Mock of the TimerHandle returned by timer.set().

Observability¶

log namespace¶

messages instance-attribute ¶

debug ¶

info ¶

warn ¶

warning ¶

error ¶

clear ¶

cache namespace¶

fetch async ¶

store async ¶

has_cache ¶

delete async ¶

exists async ¶

list_push async ¶

list_pop_all async ¶

list_len async ¶

list_len_sum async ¶

expire async ¶

set_data ¶

clear ¶

cdr namespace¶

enabled property ¶

typed_records property ¶

session_key staticmethod ¶

finalize ¶

write ¶

clear ¶

The record a collector receives¶

Writing to several sinks¶

metrics namespace¶

counter ¶

gauge ¶

histogram ¶

clear ¶

Counter¶

inc ¶

labels ¶

Gauge¶

inc ¶

dec ¶

set ¶

labels ¶

Histogram¶

observe ¶

labels ¶

timer namespace¶

scheduled property ¶

every ¶

set ¶

cancel ¶

fire ¶

TimerHandle¶

`log` namespace¶

messages `instance-attribute` ¶

`cache` namespace¶

fetch `async` ¶

store `async` ¶

delete `async` ¶

exists `async` ¶

list_push `async` ¶

list_pop_all `async` ¶

list_len `async` ¶

list_len_sum `async` ¶

expire `async` ¶

`cdr` namespace¶

enabled `property` ¶

typed_records `property` ¶

session_key `staticmethod` ¶

`metrics` namespace¶

`Counter`¶

`Gauge`¶

`Histogram`¶

`timer` namespace¶

scheduled `property` ¶

`TimerHandle`¶