You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
84 lines
2.5 KiB
Python
84 lines
2.5 KiB
Python
"""Tests: _run_pipeline halts on first failure and returns False."""
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
def _make_mock_proc(returncode: int) -> AsyncMock:
|
|
proc = AsyncMock()
|
|
proc.returncode = returncode
|
|
proc.communicate = AsyncMock(return_value=(b"", None))
|
|
return proc
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pipeline_halts_on_first_failure():
|
|
"""If first subprocess fails, downstream commands must not run."""
|
|
from apps.web.paper_trading_service import AutoScheduler
|
|
|
|
sched = AutoScheduler()
|
|
sched._log = lambda msg: None
|
|
sched._dry_run = False
|
|
|
|
create_calls = []
|
|
|
|
async def fake_create_subprocess(*args, **kwargs):
|
|
create_calls.append(args[0])
|
|
# First call fails, others would succeed
|
|
rc = 1 if len(create_calls) == 1 else 0
|
|
return _make_mock_proc(rc)
|
|
|
|
with patch("asyncio.create_subprocess_exec", side_effect=fake_create_subprocess):
|
|
cmds = [["python", "-m", "step1"], ["python", "-m", "step2"], ["python", "-m", "step3"]]
|
|
ok = await sched._run_pipeline(cmds)
|
|
|
|
assert ok is False
|
|
# Only first command ran
|
|
assert len(create_calls) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pipeline_returns_true_on_all_success():
|
|
"""All steps succeed → returns True."""
|
|
from apps.web.paper_trading_service import AutoScheduler
|
|
|
|
sched = AutoScheduler()
|
|
sched._log = lambda msg: None
|
|
sched._dry_run = False
|
|
|
|
async def fake_create_subprocess(*args, **kwargs):
|
|
return _make_mock_proc(0)
|
|
|
|
with patch("asyncio.create_subprocess_exec", side_effect=fake_create_subprocess):
|
|
cmds = [["python", "-m", "step1"], ["python", "-m", "step2"]]
|
|
ok = await sched._run_pipeline(cmds)
|
|
|
|
assert ok is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pipeline_exception_treated_as_failure():
|
|
"""If create_subprocess_exec raises, pipeline returns False and stops."""
|
|
from apps.web.paper_trading_service import AutoScheduler
|
|
|
|
sched = AutoScheduler()
|
|
sched._log = lambda msg: None
|
|
sched._dry_run = False
|
|
call_count = 0
|
|
|
|
async def fake_create_subprocess(*args, **kwargs):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
if call_count == 1:
|
|
raise OSError("boom")
|
|
return _make_mock_proc(0)
|
|
|
|
with patch("asyncio.create_subprocess_exec", side_effect=fake_create_subprocess):
|
|
cmds = [["python", "-m", "step1"], ["python", "-m", "step2"]]
|
|
ok = await sched._run_pipeline(cmds)
|
|
|
|
assert ok is False
|
|
assert call_count == 1
|