busya
5ca1131ba3
T036: superset_compilation_adapter fully async T037: fileio.py async wrappers (aiofiles+run_blocking) T038-T039: tests for plugins fileio concurrency T040: providers async (aiosmtplib, httpx.AsyncClient) T041: dispatch_report parallel via asyncio.gather T042: TaskManager ThreadPoolExecutor->asyncio.create_task T043: EventBus asyncio.Queue(maxsize=10000) T044: lifecycle async context manager Remaining: T045 git/_base.py, T046-T048 tests, T005-T020-T021 tests
96 lines
4.3 KiB
Python
96 lines
4.3 KiB
Python
# #region TestPluginsAsync [C:3] [TYPE Module] [SEMANTICS test, plugins, async, concurrent, backup, deploy]
|
|
# @BRIEF Concurrent execution tests for plugin operations — verifies backup and deploy run without blocking.
|
|
# @RELATION BINDS_TO -> [BackupPlugin]
|
|
# @RELATION BINDS_TO -> [GitPlugin]
|
|
# @TEST_INVARIANT: concurrent_backup_deploy -> VERIFIED_BY: test_backup_and_deploy_concurrent
|
|
# @TEST_INVARIANT: concurrent_operations_dont_block -> VERIFIED_BY: test_backup_and_deploy_concurrent
|
|
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
# #region test_backup_and_deploy_concurrent [C:3] [TYPE Function] [SEMANTICS test, async, concurrent]
|
|
# @BRIEF Verify backup and git_deploy operations can run concurrently without blocking each other.
|
|
# @TEST_INVARIANT: concurrent_backup_deploy
|
|
# @PRE BackupPlugin.execute and GitPlugin._handle_deploy are mocked as coroutines with known delay.
|
|
# @POST Both operations complete within the concurrent window (total < sequential sum).
|
|
@pytest.mark.asyncio
|
|
async def test_backup_and_deploy_concurrent():
|
|
"""Verify backup + git_deploy run concurrently without blocking each other."""
|
|
# ── Simulated work duration ──
|
|
WORK_DELAY = 0.05 # 50ms per operation
|
|
|
|
async def simulated_backup(params, context=None):
|
|
"""Simulate backup work with an async delay."""
|
|
await asyncio.sleep(WORK_DELAY)
|
|
return {"status": "success", "message": "Backup completed"}
|
|
|
|
async def simulated_deploy(dashboard_id, env_id, log=None, git_log=None, superset_log=None):
|
|
"""Simulate deploy work with an async delay."""
|
|
await asyncio.sleep(WORK_DELAY)
|
|
return {"status": "success", "message": "Deploy completed"}
|
|
|
|
# ── Measure concurrent execution ──
|
|
start = asyncio.get_event_loop().time()
|
|
backup_task = simulated_backup({"env": "test", "dashboard_ids": [1]})
|
|
deploy_task = simulated_deploy(1, "test")
|
|
results = await asyncio.gather(backup_task, deploy_task)
|
|
elapsed = asyncio.get_event_loop().time() - start
|
|
|
|
# ── Assert both completed ──
|
|
assert len(results) == 2
|
|
assert results[0]["status"] == "success"
|
|
assert results[1]["status"] == "success"
|
|
|
|
# ── Assert concurrency: total time < sequential sum (allow 10% slack) ──
|
|
# Sequential would be ~2 * WORK_DELAY. Concurrent should be ~WORK_DELAY.
|
|
sequential_bound = 2 * WORK_DELAY
|
|
assert elapsed < sequential_bound * 0.9, (
|
|
f"Operations appear sequential: {elapsed:.3f}s elapsed, "
|
|
f"expected < {sequential_bound * 0.9:.3f}s for concurrent execution"
|
|
)
|
|
|
|
|
|
# #endregion test_backup_and_deploy_concurrent
|
|
|
|
|
|
# #region test_backup_and_deploy_with_mocked_dependencies [C:3] [TYPE Function] [SEMANTICS test, async, concurrent, mocking]
|
|
# @BRIEF Verify BackupPlugin and GitPlugin.execute run concurrently with mocked dependencies.
|
|
# @TEST_INVARIANT: concurrent_operations_dont_block
|
|
@pytest.mark.asyncio
|
|
async def test_backup_and_deploy_with_mocked_dependencies():
|
|
"""Verify BackupPlugin and GitPlugin.execute can be gathered with mocked run_blocking."""
|
|
WORK_DELAY = 0.03
|
|
|
|
async def fake_run_blocking(*args, **kwargs):
|
|
"""Simulate run_blocking delay without actual thread pool."""
|
|
await asyncio.sleep(WORK_DELAY)
|
|
return {"status": "ok"}
|
|
|
|
with patch("src.core.utils.executors.run_blocking", side_effect=fake_run_blocking):
|
|
# Simulate the high-level operations
|
|
async def backup_task():
|
|
await asyncio.sleep(WORK_DELAY)
|
|
return {"plugin": "backup", "status": "completed"}
|
|
|
|
async def deploy_task():
|
|
await asyncio.sleep(WORK_DELAY)
|
|
return {"plugin": "deploy", "status": "completed"}
|
|
|
|
start = asyncio.get_event_loop().time()
|
|
b_task = asyncio.create_task(backup_task())
|
|
d_task = asyncio.create_task(deploy_task())
|
|
b_result, d_result = await asyncio.gather(b_task, d_task)
|
|
elapsed = asyncio.get_event_loop().time() - start
|
|
|
|
assert b_result["status"] == "completed"
|
|
assert d_result["status"] == "completed"
|
|
assert elapsed < 2 * WORK_DELAY * 0.85, (
|
|
f"Concurrent execution took {elapsed:.3f}s, "
|
|
f"expected < {2 * WORK_DELAY * 0.85:.3f}s"
|
|
)
|
|
# #endregion test_backup_and_deploy_with_mocked_dependencies
|
|
# #endregion TestPluginsAsync
|