pub/lct-hack
Watch
1
0
Fork
You've already forked lct-hack
 0
Code Releases Packages 5 Activity

test: смоук доказывает разовую проверку логина на B по журналу, время от создания учёток

Browse source
This commit is contained in:
GGlamer 2026-09-27 22:29:57 +03:00
commit 584f86f796
1 changed files with 11 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

15
scripts/test_cluster_failover.py
View file

@ -209,7 +209,7 @@ async def main(args: argparse.Namespace) -> int:
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool from sqlalchemy.pool import NullPool
from app.api.auth import hash_password from app.api.auth import hash_password, login_log_marker
from app.db.models import AuditLog, Session, Trainee, User from app.db.models import AuditLog, Session, Trainee, User
database_url = ( database_url = (
@ -250,19 +250,26 @@ async def main(args: argparse.Namespace) -> int:
trainee_id=trainee.id, blocked=False), trainee_id=trainee.id, blocked=False),
]) ])
await db.commit() await db.commit()
users_committed_at = time.monotonic()
instructor_cookie = login(backend, instructor_login, instructor_password) instructor_cookie = login(backend, instructor_login, instructor_password)
login_completed_at = time.monotonic()
trainee_cookie = login(backend, trainee_login, trainee_password) trainee_cookie = login(backend, trainee_login, trainee_password)
control_sent_at = await send_control(ws_base, session_id, instructor_cookie, { control_sent_at = await send_control(ws_base, session_id, instructor_cookie, {
"type": "scenario.start", "scenario_id": args.scenario, "type": "scenario.start", "scenario_id": args.scenario,
"trainee": trainee.name, "trainee_id": str(trainee.id), "trainee": trainee.name, "trainee_id": str(trainee.id),
"mode": "training", "exercise": "dds", "mode": "training", "exercise": "dds",
}) })
result["login_to_control_seconds"] = round(control_sent_at - login_completed_at, 4) result["users_commit_to_control_seconds"] = round(control_sent_at - users_committed_at, 4)
if args.expect_initial_owner == "backend-b": if args.expect_initial_owner == "backend-b":
result["checks"]["control_sent_within_first_second"] = ( result["checks"]["control_sent_within_first_second"] = (
result["login_to_control_seconds"] < 1 result["users_commit_to_control_seconds"] < 1
)
# Сверка B идёт раз в секунду в случайной фазе, и время само по себе
# не доказывает, что B не знал логин. Доказательство — запись B о
# разовой проверке именно этого логина.
result["checks"]["backend_b_resolved_login_one_shot"] = (
login_log_marker(instructor_login)
in compose(args, "logs", "--no-color", "backend-b")
) )
async def session_row(): async def session_row():