mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-12 03:48:56 +00:00
9b53ee878f
* perf(ci): distribute unit cases and reuse security sweep connections * revert(ci): reject first speed trial after new failures * perf(ci): avoid repeated test discovery and parsing * perf(ci): isolate parallel security workers * revert(ci): restore measured security execution isolation * perf(tests): reuse source scans and deduplicate cache cleanup * perf(ci): restore parallel security with safe log batching * test(ci): isolate fixture state and schema comparisons * test(ci): make schema and scheduler fixtures deterministic * test(ci): drain queued logging before fixture teardown * test(ci): preserve logging delivery without idle worker restarts * test(ci): keep guardrail sinks available during parallel bursts * test(ci): isolate full proxy lifecycle from gateway imports * test(ci): preserve first failure responses and sweep evidence * test(ci): restore documentation URL settings between cases * test(ci): isolate open MCP tools in private gateways * test(ci): group private MCP cases and handle exited workers * test(ci): retain proxy exits and MCP failure details * test(ci): repair merge interactions with main Key the schema shard's full clone on artifact-name, since main removed the unit-flag input. Give each owned Redis its own log file again. Hand the realtime transcription tests main added a logging object whose dispatch handlers are awaitable, so the per-test logging drain no longer waits on a MagicMock task forever * perf(ci): shard owned integration cases and close lifecycle leaks * fix(ci): qualify shard execution and harden interrupted cleanup * fix(ci): bound worker supervision and own fault traffic * perf(ci): reuse unparametrized fixtures across bounded case groups * fix(ci): preserve boot fault allowances and complete exit records * test(ci): tolerate startup retries and retain readiness details * perf(ci): balance integration shards by recorded duration * fix(ci): load migration database helpers only for workers
180 lines
7.9 KiB
Python
180 lines
7.9 KiB
Python
from __future__ import annotations
|
|
|
|
from itertools import groupby
|
|
from types import MappingProxyType
|
|
from typing import Final
|
|
|
|
import pytest
|
|
|
|
from tests.integration._support.ordering import (
|
|
CollectedCase,
|
|
ScopedParameter,
|
|
case_shards,
|
|
fixture_groups,
|
|
parse_timings,
|
|
)
|
|
from tests.integration.conftest import _order_key
|
|
|
|
|
|
def test_seeded_order_reuses_each_scoped_configuration_without_dropping_cases() -> None:
|
|
values: Final = (
|
|
"D1",
|
|
"D1",
|
|
"D1",
|
|
"D1",
|
|
"D1",
|
|
"D2",
|
|
"D2",
|
|
"D2",
|
|
"D2",
|
|
"D2",
|
|
"D3",
|
|
"D3",
|
|
"D3",
|
|
"D3",
|
|
"D4",
|
|
"D4",
|
|
"D4",
|
|
"D4",
|
|
)
|
|
cases: Final = tuple(
|
|
CollectedCase(
|
|
f"tests/integration/security/test_slots.py::test_route[{value}-{index}]",
|
|
(ScopedParameter("module:slots", "rig", value),),
|
|
)
|
|
for index, value in enumerate(values)
|
|
)
|
|
groups: Final = fixture_groups(cases)
|
|
ordered: Final = tuple(sorted(cases, key=lambda case: _order_key(1848224111, case.nodeid, groups[case.nodeid])))
|
|
configurations: Final = tuple(value for value, _ in groupby(case.parameters[0].value for case in ordered))
|
|
assert len(configurations) == len(set(values))
|
|
assert set(configurations) == set(values)
|
|
assert frozenset(case.nodeid for case in ordered) == frozenset(case.nodeid for case in cases)
|
|
assert len(ordered) == len(cases)
|
|
assert ordered == tuple(sorted(cases, key=lambda case: _order_key(1848224111, case.nodeid, groups[case.nodeid])))
|
|
assert ordered != tuple(sorted(cases, key=lambda case: _order_key(984872314, case.nodeid, groups[case.nodeid])))
|
|
|
|
|
|
def test_equal_unhashable_parameters_share_a_group_only_within_their_fixture_owner() -> None:
|
|
first: Final = CollectedCase("test_slots.py::test_first", (ScopedParameter("class:first", "rig", {"slot": "D1"}),))
|
|
equal: Final = CollectedCase("test_slots.py::test_equal", (ScopedParameter("class:first", "rig", {"slot": "D1"}),))
|
|
other_owner: Final = CollectedCase(
|
|
"test_slots.py::test_other", (ScopedParameter("class:second", "rig", {"slot": "D1"}),)
|
|
)
|
|
other_file: Final = CollectedCase("test_other.py::test_first", first.parameters)
|
|
plain: Final = CollectedCase("test_slots.py::test_plain")
|
|
groups: Final = fixture_groups((first, equal, other_owner, other_file, plain))
|
|
assert groups[first.nodeid] == groups[equal.nodeid]
|
|
assert groups[first.nodeid] != groups[other_owner.nodeid]
|
|
assert groups[first.nodeid] != groups[other_file.nodeid]
|
|
assert groups[plain.nodeid] == ""
|
|
|
|
|
|
def test_shards_balance_cases_without_splitting_a_scoped_configuration() -> None:
|
|
cases: Final = tuple(
|
|
CollectedCase(
|
|
f"test_slots.py::test_route[{index}]",
|
|
(ScopedParameter("module:slots", "rig", index // 4),),
|
|
("module:slots:extra",) if index % 2 else (),
|
|
)
|
|
for index in range(20)
|
|
)
|
|
plain: Final = tuple(CollectedCase(f"test_slots.py::test_plain[{index}]") for index in range(7))
|
|
shards: Final = case_shards(cases + plain, 3)
|
|
assert frozenset.union(*shards) == frozenset(case.nodeid for case in cases + plain)
|
|
assert sum(len(shard) for shard in shards) == len(cases + plain)
|
|
assert max(map(len, shards)) - min(map(len, shards)) <= 1
|
|
for _, group in groupby(cases, key=lambda case: case.parameters[0].value):
|
|
configuration: Final = frozenset(case.nodeid for case in group)
|
|
assert sum(configuration <= shard for shard in shards) == 1
|
|
assert shards == case_shards(tuple(reversed(cases + plain)), 3)
|
|
|
|
|
|
def test_shards_reuse_unparametrized_fixtures_and_keep_different_results_separate() -> None:
|
|
shared: Final = tuple(
|
|
CollectedCase(f"test_sockets.py::test_override[{index}]", fixtures=("module:test_sockets.py:override",))
|
|
for index in range(4)
|
|
)
|
|
other: Final = CollectedCase("test_sockets.py::test_default", fixtures=("module:test_sockets.py:default",))
|
|
plain: Final = tuple(CollectedCase(f"test_other.py::test_plain[{index}]") for index in range(7))
|
|
shards: Final = case_shards((*shared, other, *plain), 3)
|
|
configuration: Final = frozenset(case.nodeid for case in shared)
|
|
assert sum(configuration <= shard for shard in shards) == 1
|
|
assert sorted(map(len, shards)) == [4, 4, 4]
|
|
assert frozenset.union(*shards) == frozenset(case.nodeid for case in (*shared, other, *plain))
|
|
groups: Final = fixture_groups((*shared, other))
|
|
assert len({groups[case.nodeid] for case in shared}) == 1
|
|
assert groups[other.nodeid] != groups[shared[0].nodeid]
|
|
assert shards == case_shards(tuple(reversed((*shared, other, *plain))), 3)
|
|
|
|
|
|
def test_an_oversized_unparametrized_fixture_is_split_into_bounded_groups() -> None:
|
|
cases: Final = tuple(
|
|
CollectedCase(f"test_sockets.py::test_route[{index:02}]", fixtures=("module:test_sockets.py:owned",))
|
|
for index in range(20)
|
|
)
|
|
shards: Final = case_shards(cases, 4)
|
|
assert tuple(map(len, shards)) == (5, 5, 5, 5)
|
|
assert frozenset.union(*shards) == frozenset(case.nodeid for case in cases)
|
|
assert sum(len(shard) for shard in shards) == len(cases)
|
|
|
|
|
|
def test_measured_cost_splits_an_expensive_shared_fixture_without_losing_cases() -> None:
|
|
expensive: Final = tuple(
|
|
CollectedCase(f"test_tracing.py::test_route[{index}]", fixtures=("module:test_tracing.py:gateway",))
|
|
for index in range(4)
|
|
)
|
|
cheap: Final = tuple(CollectedCase(f"test_other.py::test_route[{index}]") for index in range(8))
|
|
cases: Final = (*expensive, *cheap)
|
|
durations: Final = MappingProxyType({case.nodeid: 40.0 if case in expensive else 1.0 for case in cases})
|
|
shards: Final = case_shards(cases, 4, durations)
|
|
assert sorted(sum(durations[nodeid] for nodeid in shard) for shard in shards) == [42.0] * 4
|
|
assert frozenset.union(*shards) == frozenset(case.nodeid for case in cases)
|
|
assert sum(len(shard) for shard in shards) == len(cases)
|
|
assert shards == case_shards(tuple(reversed(cases)), 4, durations)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"content",
|
|
(
|
|
"invalid",
|
|
"{}",
|
|
'[{"nodeid":"case","seconds":-1}]',
|
|
'[{"nodeid":"case","seconds":"1"}]',
|
|
'[{"nodeid":"case","seconds":NaN}]',
|
|
'[{"nodeid":"case","seconds":Infinity}]',
|
|
'[{"nodeid":"case","seconds":1},{"nodeid":"case","seconds":2}]',
|
|
),
|
|
)
|
|
def test_invalid_timing_cache_preserves_the_complete_cold_partition(content: str) -> None:
|
|
cases: Final = tuple(CollectedCase(f"test_owned.py::test_route[{index}]") for index in range(9))
|
|
durations: Final = parse_timings(content)
|
|
assert not durations
|
|
assert case_shards(cases, 3, durations) == case_shards(cases, 3)
|
|
|
|
|
|
def test_timing_cache_keeps_zero_duration_skips_and_assigns_new_cases() -> None:
|
|
durations: Final = parse_timings('[{"nodeid":"skip","seconds":0},{"nodeid":"old","seconds":12}]')
|
|
cases: Final = (CollectedCase("skip"), CollectedCase("new"))
|
|
shards: Final = case_shards(cases, 2, durations)
|
|
assert durations["skip"] > 0
|
|
assert frozenset.union(*shards) == frozenset({"skip", "new"})
|
|
assert tuple(map(len, shards)) == (1, 1)
|
|
|
|
|
|
def test_measured_cost_keeps_parametrized_fixture_configurations_atomic() -> None:
|
|
shared: Final = tuple(
|
|
CollectedCase(
|
|
f"test_tracing.py::test_route[{index}]",
|
|
(ScopedParameter("module:test_tracing.py", "configuration", "shared"),),
|
|
)
|
|
for index in range(3)
|
|
)
|
|
plain: Final = tuple(CollectedCase(f"test_other.py::test_route[{index}]") for index in range(5))
|
|
cases: Final = (*shared, *plain)
|
|
durations: Final = MappingProxyType({case.nodeid: 40.0 if case in shared else 1.0 for case in cases})
|
|
shards: Final = case_shards(cases, 3, durations)
|
|
configuration: Final = frozenset(case.nodeid for case in shared)
|
|
assert sum(configuration <= shard for shard in shards) == 1
|
|
assert frozenset.union(*shards) == frozenset(case.nodeid for case in cases)
|
|
assert sum(len(shard) for shard in shards) == len(cases)
|