xsl
93b88db2f5
Covers routing, failover, the circuit breaker, structured-output retries, the cache-prefix validator, cost computation including cached-token pricing, the daily budget breaker, the quota protocol, the task state machine and log redaction. HTTP is mocked -- no unit test spends money. The integration test runs the real slice against PostgreSQL and Redis with a faked provider.
56 lines
1.6 KiB
Python
Executable File
56 lines
1.6 KiB
Python
Executable File
"""Snowflake ids (架构文档 §6.1 原则 2)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from packages.domain.ids import (
|
|
DEFAULT_EPOCH_MS,
|
|
SnowflakeGenerator,
|
|
next_id,
|
|
timestamp_ms_of,
|
|
)
|
|
|
|
|
|
def test_ids_are_unique_and_monotonic() -> None:
|
|
generator = SnowflakeGenerator(node_id=1)
|
|
ids = [generator.next_id() for _ in range(5000)]
|
|
assert len(set(ids)) == len(ids)
|
|
assert ids == sorted(ids)
|
|
|
|
|
|
def test_ids_fit_in_a_signed_64_bit_integer() -> None:
|
|
assert 0 < next_id() < 2**63
|
|
|
|
|
|
def test_different_nodes_do_not_collide() -> None:
|
|
a = SnowflakeGenerator(node_id=1)
|
|
b = SnowflakeGenerator(node_id=2)
|
|
assert not {a.next_id() for _ in range(500)} & {b.next_id() for _ in range(500)}
|
|
|
|
|
|
def test_node_id_is_bounded() -> None:
|
|
with pytest.raises(ValueError, match="node_id"):
|
|
SnowflakeGenerator(node_id=1024)
|
|
|
|
|
|
def test_timestamp_is_recoverable_for_partition_pruning() -> None:
|
|
generator = SnowflakeGenerator(node_id=3)
|
|
recovered = timestamp_ms_of(generator.next_id())
|
|
assert recovered > DEFAULT_EPOCH_MS
|
|
|
|
|
|
def test_sequence_rollover_within_one_millisecond() -> None:
|
|
"""More than 4096 ids in a single ms must wait, not duplicate."""
|
|
frozen = DEFAULT_EPOCH_MS + 1_000
|
|
calls = {"n": 0}
|
|
|
|
def fake_now() -> int:
|
|
calls["n"] += 1
|
|
# Hold the clock for the first 5000 calls, then let it advance.
|
|
return frozen if calls["n"] < 5000 else frozen + 1
|
|
|
|
generator = SnowflakeGenerator(node_id=1, clock=fake_now)
|
|
ids = [generator.next_id() for _ in range(4200)]
|
|
assert len(set(ids)) == len(ids)
|