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

enabled `property` ¶

enabled: bool

Whether the CDR system is enabled.

write ¶

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

Write a CDR for the given request or B2BUA call.

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 CDR was queued successfully.

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

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

write ¶

clear ¶

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

`metrics` namespace¶

`Counter`¶

`Gauge`¶

`Histogram`¶

`timer` namespace¶

scheduled `property` ¶

`TimerHandle`¶