xsl
f46ecc5eb0
Two defects that only surface against the real DeepSeek API, both of which break the cost model rather than the code: deepseek-v4-flash enables thinking mode by default. The same single-turn question took 13.5s with it on and 1.8s with it off, and the thinking tokens are billed at the output rate. That misses NFR-PERF-02b (15s per turn) and inflates every conversational call. SceneConfig now carries an explicit `thinking` flag, all four V1 scenes set it to false, and the provider always sends the parameter rather than relying on the upstream default. Reasoning tokens are recorded separately in Usage.tokens_reasoning so the cost of ever turning it back on is visible. DeepSeek moves to peak/off-peak pricing on 2026-08-17, with peak (Beijing 09:00-12:00 and 14:00-18:00) at twice the off-peak rate. Reports are generated mostly during peak hours, so a single blended rate would leave ob_ai_call -- the only input to V2 pricing -- off by nearly a factor of two. PeakAwareRateTable resolves the rate at call time; cost is computed and stored then, never recomputed later at whatever the current rate happens to be. Also here: - Replace the live API key in the redaction test with a synthetic one. The test that proves we redact secrets was itself a secret leak waiting for its first commit. - Correct LLM_COST_RATES_JSON, which held placeholder rates. - Make ids.SnowflakeGenerator take an injectable clock so rollover and clock-rollback are testable without patching a slots class. - Rename exceptions to the -Error suffix (N818) and clear the remaining ruff, mypy and import-linter findings, including two dead lines: a duplicate pytestmark assignment and a no-op assert on a None return. Gates: ruff, mypy (96 files incl. tests), 4/4 import contracts, detect-secrets, 166 unit + 3 integration tests, migration up/down round-trip.
64 lines
2.0 KiB
Python
Executable File
64 lines
2.0 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_clock() -> 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_clock)
|
|
ids = [generator.next_id() for _ in range(4200)]
|
|
assert len(set(ids)) == len(ids)
|
|
|
|
|
|
def test_clock_moving_backwards_never_yields_duplicates() -> None:
|
|
"""时钟回拨（NTP 校正）时必须继续单调，不能重发已发过的 ID。"""
|
|
times = iter([DEFAULT_EPOCH_MS + 500, DEFAULT_EPOCH_MS + 400, DEFAULT_EPOCH_MS + 400])
|
|
generator = SnowflakeGenerator(node_id=1, clock=lambda: next(times))
|
|
ids = [generator.next_id() for _ in range(3)]
|
|
assert len(set(ids)) == 3
|