loop-uh
f6e48891f9
All checks were successful
Desktop source guards / guards (push) Successful in 5s
Скрипт выпуска и CI запускают каждый гард из release_guards.txt как
`python <файл>` и смотрят только на код возврата. Семь файлов не вызывали
свои test_* вовсе и проходили, ничего не проверив, а в трёх ручной список
вызовов отстал на один новый тест. Теперь каждый гард вызывает все свои
test_* сам, тем же блоком, что уже стоял в test_web_proxy_carrier_flow.py,
а новый гард test_release_guards_run_their_tests.py следит, чтобы так
оставалось у всех файлов списка.
Под запуском всплыли устаревшие проверки, сломанные коммитом 744109cc78
(«Pace MTProxy dials instead of steering them»):
- он намеренно убрал подстройку MTProxy под кэш возможностей и
принудительный ChromeModern: профиль ClientHello остаётся тем, что выбрал
пользователь. Проверки в test_mtproxy_target_flow.py и
test_proxy_capability_cache.py требовали старое поведение и теперь
требуют новое: в ветке MTProxy нет обращения к кэшу, а
BoringMtproxyStealthOptions сохраняет выбранный профиль;
- reportMtproxySuccess удалён, поэтому проверка «первые данные отмечаются
один раз на сокет» смотрит на то, что сейчас стоит под этой защитой:
прогресс, событие диагностики и noteDataPathSuccess;
- память маршрутов он оставил, но код переписан: проверка порядка
(ответившие адреса, затем неопробованные, затем неудачные) идёт по
нынешнему виду routeOrder.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
83 lines
3.1 KiB
Python
83 lines
3.1 KiB
Python
from pathlib import Path
|
|
from session_private_sources import read_session_private_sources
|
|
|
|
|
|
SOURCE_DIR = Path(__file__).resolve().parents[1]
|
|
ROOT = SOURCE_DIR.parents[1]
|
|
PROXY_DIR = SOURCE_DIR / "mtproto" / "proxy"
|
|
CONTROL_H = PROXY_DIR / "control_plane.h"
|
|
CONTROL_CPP = PROXY_DIR / "control_plane.cpp"
|
|
TLS_CPP = PROXY_DIR / "mtproxy" / "tls_socket.cpp"
|
|
RESOLVING_CPP = PROXY_DIR / "resolving_connection.cpp"
|
|
|
|
|
|
def read(path):
|
|
assert path.exists(), f"missing expected source file: {path}"
|
|
return path.read_text(encoding="utf-8")
|
|
|
|
|
|
def test_raw_reducer_rejects_probe_and_stale_generation_facts():
|
|
source = read(CONTROL_CPP)
|
|
reduce = function_body(
|
|
source, "ProxyConnectionStatus ProxyControlPlane::Reduce(")
|
|
update = function_body(source, "ProxyConnectionStatus ApplySelectedStatusUpdate(")
|
|
|
|
assert "IsProxyCheck(fact.status.attempt.use)" in reduce
|
|
assert "return current;" in reduce.split(
|
|
"IsProxyCheck(fact.status.attempt.use)", 1)[1].split("}", 1)[0]
|
|
assert "IsOlderProxyGeneration(current.attempt, update.attempt)" in update
|
|
assert "IsOlderAttempt(current.attempt, update.attempt)" in update
|
|
assert "RelaySuccessIsFresh(current)" in update
|
|
assert "kFreshRelaySuccessWindow" in source
|
|
# An mtproxy carries every session at once, so a single socket failing
|
|
# while the others relay must not repaint the shield.
|
|
assert "ShadowedByFreshRelaySuccess(current, fact)" in reduce
|
|
|
|
|
|
def test_tls_socket_owns_serverhello_timeout_and_one_terminal_path():
|
|
source = read(TLS_CPP)
|
|
timeout = function_body(source, "void TlsSocket::handleServerHelloTimeout()")
|
|
terminal = function_body(source, "bool TlsSocket::finishTerminal(")
|
|
|
|
assert "_serverHelloDeadline" in timeout
|
|
assert "_serverHelloDeadline - now" in timeout
|
|
assert "finishTerminal(" in timeout
|
|
assert "if (_terminal)" in terminal
|
|
assert "_terminal = true;" in terminal
|
|
assert "_serverHelloTimer.cancel();" in terminal
|
|
assert "_terminalFailure = collectTransportFailure();" in terminal
|
|
assert "reportTransportEvent(" in terminal
|
|
|
|
|
|
def test_resolving_wrapper_keeps_typed_route_and_full_cycle_boundaries():
|
|
source = read(RESOLVING_CPP)
|
|
budget = function_body(
|
|
source, "crl::time ResolvingConnection::fullConnectTimeout() const")
|
|
|
|
assert "TypedRouteFailure(" in source
|
|
assert "MergeExhaustedFailure(" in source
|
|
assert "_mtproxyPlan.serverHelloTimeout" in budget
|
|
assert "kRouteAttemptTimeout" in budget
|
|
assert "kOnlyRouteAttemptTimeout" in budget
|
|
assert "kFullConnectTimeoutSafetyMargin" in budget
|
|
|
|
|
|
def function_body(text, signature):
|
|
start = text.index(signature)
|
|
brace = text.index("{", start)
|
|
depth = 0
|
|
for index in range(brace, len(text)):
|
|
if text[index] == "{":
|
|
depth += 1
|
|
elif text[index] == "}":
|
|
depth -= 1
|
|
if depth == 0:
|
|
return text[brace + 1:index]
|
|
raise AssertionError(f"function body not found: {signature}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
for name, value in list(globals().items()):
|
|
if name.startswith("test_") and callable(value):
|
|
value()
|
|
print("ok")
|