PlantAI-dev/tests/unit/test_ppi_batch_owner_isolation.py

from __future__ import annotations

import json
import subprocess
from pathlib import Path

import pytest


# 主分支已完成 PPI/KaKs 修复，index.html 中 ppiKaksTemporaryCircuitBreakerEnabled 已置为 false。
# 以下用例仍假设临时熔断处于开启态，需按新状态重写后再启用。
_circuit_breaker_disabled = pytest.mark.skip(
    reason="临时熔断已随 PPI/KaKs 修复关闭，用例待按新状态重写"
)


def _source_block(source: str, start: str, end: str) -> str:
    start_index = source.index(start)
    return source[start_index:source.index(end, start_index)]


def _run_node(source: str) -> dict:
    completed = subprocess.run(
        ["/usr/bin/node", "-e", source],
        check=True,
        capture_output=True,
        text=True,
    )
    return json.loads(completed.stdout)


def _index_html() -> str:
    return Path("apps/web/vue3/index.html").read_text(encoding="utf-8")


def test_joint_owner_preflight_continues_after_one_owner_pauses() -> None:
    helper = _source_block(
        _index_html(),
        "const preflightPpiKaksDependencyOwners = async",
        "const ensureOrthologProducerForPpiKaks = async",
    )
    result = _run_node(
        helper
        + """
(async () => {
  const calls = [];
  const result = await preflightPpiKaksDependencyOwners(
    ['kaks', 'ppi'],
    async owner => {
      calls.push(owner);
      return owner === 'ppi';
    }
  );
  console.log(JSON.stringify({calls, result}));
})();
"""
    )

    assert result == {
        "calls": ["kaks", "ppi"],
        "result": [
            {"ownerStageId": "kaks", "status": "paused"},
            {"ownerStageId": "ppi", "status": "ready"},
        ],
    }


def test_ppi_kaks_pause_only_claims_owned_ortholog() -> None:
    helper = _source_block(
        _index_html(),
        "const ppiKaksOwnedOrthologOwners =",
        "const ppiKaksPauseStageIds =",
    )
    result = _run_node(
        helper
        + """
const explicit = ppiKaksOwnedOrthologOwners(
  {id: 'ortholog', params: {autoInsertedFor: ['kaks']}},
  {intentStages: ['ortholog', 'kaks']}
);
const plannerInserted = ppiKaksOwnedOrthologOwners(
  {id: 'ortholog', params: {}},
  {intentStages: ['kaks']}
);
const ppiOwned = ppiKaksOwnedOrthologOwners(
  {id: 'ortholog', params: {dependencyOwner: 'ppi'}},
  {intentStages: ['ppi']}
);
const unknown = ppiKaksOwnedOrthologOwners(
  {id: 'ortholog', params: {}},
  {}
);
console.log(JSON.stringify({explicit, plannerInserted, ppiOwned, unknown}));
"""
    )

    assert result == {
        "explicit": [],
        "plannerInserted": ["kaks"],
        "ppiOwned": ["ppi"],
        "unknown": [],
    }


def test_joint_owner_pauses_are_stored_without_overwriting_each_other() -> None:
    helper = _source_block(
        _index_html(),
        "const ppiKaksOwnedOrthologOwners =",
        "const mergePpiReferenceSpeciesIntoBatch =",
    )
    result = _run_node(
        """
const batchExecuteParams = {value: {intentStages: ['kaks', 'ppi']}};
const batchExecuteStages = {value: [
  {id: 'ortholog', params: {autoInsertedFor: ['kaks', 'ppi']}},
  {id: 'kaks'}, {id: 'ppi'}
]};
const stageExecutionStatus = {value: {ortholog: 'pending', kaks: 'pending', ppi: 'pending'}};
const stagePauseReasons = {};
const getStageId = stage => typeof stage === 'object' ? stage.id : stage;
const recordStagePause = (stageId, pause) => { stagePauseReasons[stageId] = pause.reason; };
"""
        + helper
        + """
storePpiKaksReferencePause(['kaks'], {issue_code: 'missing_cds'}, 'KaKs paused');
applyStoredPpiKaksPauseStatuses();
const afterKaks = {...stageExecutionStatus.value};
storePpiKaksReferencePause(['ppi'], {issue_code: 'missing_owner'}, 'PPI paused');
applyStoredPpiKaksPauseStatuses();
console.log(JSON.stringify({
  pauseOwners: Object.keys(batchExecuteParams.value.ppiKaksDependencyPauses).sort(),
  afterKaks,
  afterBoth: stageExecutionStatus.value,
  reasons: stagePauseReasons
}));
"""
    )

    assert result["pauseOwners"] == ["kaks", "ppi"]
    assert result["afterKaks"] == {"ortholog": "pending", "kaks": "paused", "ppi": "pending"}
    assert result["afterBoth"] == {"ortholog": "paused", "kaks": "paused", "ppi": "paused"}
    assert result["reasons"] == {
        "kaks": "missing_cds",
        "ppi": "missing_owner",
        "ortholog": "missing_cds",
    }


def test_batch_start_uses_temporary_circuit_breaker_without_owner_preflight() -> None:
    start_block = _source_block(
        _index_html(),
        "const startBatchExecute = async",
        "const executeStagesWithDependencies = async",
    )

    assert "applyPpiKaksTemporaryCircuitBreaker();" in start_block
    assert "if (ppiKaksTemporaryCircuitBreakerEnabled)" in start_block
    assert "else {\n                        await ensureOrthologProducerForPpiKaks();\n                    }" in start_block
    assert "await executeStagesWithDependencies(stages, plannedStageIds);" in start_block


@_circuit_breaker_disabled
def test_temporary_circuit_breaker_pauses_only_owned_ortholog() -> None:
    source = _index_html()
    owner_helper = _source_block(
        source,
        "const ppiKaksOwnedOrthologOwners =",
        "const ppiKaksPauseStageIds =",
    )
    policy_helper = _source_block(
        source,
        "const ppiKaksTemporaryCircuitBreakerEnabled =",
        "const batchIncludesPpiKaks =",
    )
    result = _run_node(
        "const ppiKaksDependencyStages = new Set(['ppi', 'kaks']);\n"
        "const getStageId = stage => typeof stage === 'object' ? stage.id : stage;\n"
        + owner_helper
        + policy_helper
        + """
const owned = ppiKaksTemporaryPauseStageIds(
  [{id: 'ortholog', params: {autoInsertedFor: ['ppi', 'kaks']}}, {id: 'ppi'}, {id: 'kaks'}],
  {intentStages: ['ppi', 'kaks']}
);
const explicit = ppiKaksTemporaryPauseStageIds(
  [{id: 'ortholog', params: {}}, {id: 'ppi'}],
  {intentStages: ['ortholog', 'ppi']}
);
const unrelated = ppiKaksTemporaryPauseStageIds(
  [{id: 'ortholog', params: {}}, {id: 'motif'}],
  {intentStages: ['ortholog', 'motif']}
);
const onePlannedOwner = ppiKaksTemporaryPauseStageIds(
  [{id: 'ortholog', params: {autoInsertedFor: ['ppi', 'kaks']}}, {id: 'ppi'}],
  {intentStages: ['ppi']}
);
console.log(JSON.stringify({owned, explicit, unrelated, onePlannedOwner}));
"""
    )

    assert result == {
        "owned": ["ortholog", "ppi", "kaks"],
        "explicit": ["ppi"],
        "unrelated": [],
        "onePlannedOwner": ["ortholog", "ppi"],
    }


@_circuit_breaker_disabled
def test_full_gene_family_plan_keeps_explicit_ortholog_and_other_stages_runnable() -> None:
    source = _index_html()
    owner_helper = _source_block(
        source,
        "const ppiKaksOwnedOrthologOwners =",
        "const ppiKaksPauseStageIds =",
    )
    policy_helper = _source_block(
        source,
        "const ppiKaksTemporaryCircuitBreakerEnabled =",
        "const batchIncludesPpiKaks =",
    )
    result = _run_node(
        "const ppiKaksDependencyStages = new Set(['ppi', 'kaks']);\n"
        "const getStageId = stage => typeof stage === 'object' ? stage.id : stage;\n"
        + owner_helper
        + policy_helper
        + """
const stageIds = [
  'phylogeny', 'motif', 'domain', 'structure', 'combo', 'promoter', 'tfbs',
  'synteny', 'interspecies_synteny', 'karyotype', 'pep', 'ortholog', 'kaks', 'ppi'
];
const stages = stageIds.map(id => ({id, params: {}}));
const paused = ppiKaksTemporaryPauseStageIds(stages, {intentStages: stageIds});
const runnable = stageIds.filter(stageId => !paused.includes(stageId));
console.log(JSON.stringify({paused, runnable}));
"""
    )

    assert result["paused"] == ["kaks", "ppi"]
    assert result["runnable"] == [
        "phylogeny",
        "motif",
        "domain",
        "structure",
        "combo",
        "promoter",
        "tfbs",
        "synteny",
        "interspecies_synteny",
        "karyotype",
        "pep",
        "ortholog",
    ]


@_circuit_breaker_disabled
def test_dependency_scheduler_executes_later_stages_while_ppi_kaks_are_paused() -> None:
    source = _index_html()
    owner_helper = _source_block(
        source,
        "const ppiKaksOwnedOrthologOwners =",
        "const ppiKaksPauseStageIds =",
    )
    policy_helper = _source_block(
        source,
        "const ppiKaksTemporaryCircuitBreakerEnabled =",
        "const batchIncludesPpiKaks =",
    )
    scheduler = _source_block(
        source,
        "const executeStagesWithDependencies = async",
        "const stopBatchAfterSkippedSubfamilyConfirm =",
    )
    result = _run_node(
        """
const console = {log: () => {}, error: () => {}};
const ppiKaksDependencyStages = new Set(['ppi', 'kaks']);
const getStageId = stage => typeof stage === 'object' ? stage.id : stage;
const stageDependencies = {phylogeny: [], motif: ['phylogeny'], ppi: [], kaks: []};
const shouldStopBatchExecute = {value: false};
const stageExecutionStatus = {value: {}};
const runningStages = {value: new Set()};
const maxParallelStages = 4;
const limitGwasModelHeavyWave = candidates => candidates;
const isTerminalBatchStatus = status => ['completed', 'failed', 'paused'].includes(status);
const getStageAction = stage => getStageId(stage);
const getStageLabel = stageId => stageId;
const recordStagePause = () => {};
const addAgentMessage = () => {};
const calls = [];
const executeStageWithConfirmation = async stage => {
  const stageId = getStageId(stage);
  calls.push(stageId);
  stageExecutionStatus.value[stageId] = 'completed';
};
"""
        + owner_helper
        + policy_helper
        + scheduler
        + """
const stages = [
  {id: 'ppi', dependencies: []},
  {id: 'phylogeny', dependencies: []},
  {id: 'motif', dependencies: ['phylogeny']},
  {id: 'kaks', dependencies: []}
];
const paused = ppiKaksTemporaryPauseStageIds(stages, {intentStages: stages.map(getStageId)});
stages.forEach(stage => {
  stageExecutionStatus.value[stage.id] = paused.includes(stage.id) ? 'paused' : 'pending';
});
executeStagesWithDependencies(stages, new Set(stages.map(getStageId))).then(() => {
  process.stdout.write(JSON.stringify({calls, statuses: stageExecutionStatus.value}));
});
"""
    )

    assert result == {
        "calls": ["phylogeny", "motif"],
        "statuses": {
            "ppi": "paused",
            "phylogeny": "completed",
            "motif": "completed",
            "kaks": "paused",
        },
    }


def test_submit_guard_precedes_species_preflight_and_execute_fetch() -> None:
    submit_block = _source_block(
        _index_html(),
        "const submitExecuteRequestWithAutoAnnotation = async",
        "const waitForDialogTransition = async",
    )

    guard = submit_block.index("isPpiKaksTemporarySubmissionBlocked")
    species_preflight = submit_block.index("runPpiSpeciesResolutionPreflight")
    execute_fetch = submit_block.index("fetch('/api/execute'")
    assert guard < species_preflight
    assert guard < execute_fetch
    assert "status: 'temporarily_unavailable'" in submit_block


def test_submit_guard_returns_paused_without_any_fetch() -> None:
    submit_block = _source_block(
        _index_html(),
        "const submitExecuteRequestWithAutoAnnotation = async",
        "const waitForDialogTransition = async",
    )
    result = _run_node(
        """
let fetchCalls = 0;
const fetch = async () => { fetchCalls += 1; throw new Error('fetch must not run'); };
const isPpiKaksTemporarySubmissionBlocked = () => true;
const getStageLabel = stageId => stageId === 'ppi' ? 'PPI' : stageId;
"""
        + submit_block
        + """
submitExecuteRequestWithAutoAnnotation({stage: 'ppi'}, 'ppi')
  .then(result => process.stdout.write(JSON.stringify({result, fetchCalls})));
"""
    )

    assert result["fetchCalls"] == 0
    assert result["result"] == {
        "success": False,
        "paused": True,
        "status": "temporarily_unavailable",
        "message": "PPI 功能修订中，当前暂不可用。",
    }


@_circuit_breaker_disabled
def test_submit_guard_blocks_owned_ortholog_but_not_independent_manual_ortholog() -> None:
    source = _index_html()
    owner_helper = _source_block(
        source,
        "const ppiKaksOwnedOrthologOwners =",
        "const ppiKaksPauseStageIds =",
    )
    policy_helper = _source_block(
        source,
        "const ppiKaksTemporaryCircuitBreakerEnabled =",
        "const batchIncludesPpiKaks =",
    )
    result = _run_node(
        "const ppiKaksDependencyStages = new Set(['ppi', 'kaks']);\n"
        "const getStageId = stage => typeof stage === 'object' ? stage.id : stage;\n"
        "const firstNonEmptyList = (...values) => values.find(value => Array.isArray(value) && value.length) || [];\n"
        "const batchExecuteParams = {value: {intentStages: ['kaks']}};\n"
        + owner_helper
        + policy_helper
        + """
const cases = {
  ppi: isPpiKaksTemporarySubmissionBlocked('ppi', {}),
  ownedOrtholog: isPpiKaksTemporarySubmissionBlocked(
    'ortholog', {autoInsertedFor: ['kaks']}
  ),
  independentManualOrtholog: isPpiKaksTemporarySubmissionBlocked('ortholog', {}),
};
batchExecuteParams.value.intentStages = ['ortholog', 'kaks'];
cases.explicitBatchOrtholog = isPpiKaksTemporarySubmissionBlocked(
  'ortholog', {autoInsertedFor: ['kaks']}
);
console.log(JSON.stringify(cases));
"""
    )

    assert result == {
        "ppi": True,
        "ownedOrtholog": True,
        "independentManualOrtholog": False,
        "explicitBatchOrtholog": False,
    }


def test_manual_execute_returns_before_running_or_submitting_ppi_kaks() -> None:
    manual_block = _source_block(
        _index_html(),
        "const doExecute = async",
        "const queueAutofillFromPayload = async",
    )

    guard = manual_block.index("isPpiKaksTemporarilyUnavailable(confirmParams.stage)")
    running = manual_block.index("executing.value = true")
    submit = manual_block.index("submitExecuteRequestWithAutoAnnotation")
    assert guard < running
    assert guard < submit
    assert "功能修订中，当前暂不可用" in manual_block


def test_manual_execute_shows_non_blocking_message_without_starting() -> None:
    manual_block = _source_block(
        _index_html(),
        "const doExecute = async",
        "const queueAutofillFromPayload = async",
    )
    result = _run_node(
        """
const console = {log: () => {}, error: () => {}};
const confirmParams = {stage: 'ppi'};
const confirmDialogVisible = {value: true};
const executing = {value: false};
const isExecuting = {value: false};
let timerClears = 0;
const warnings = [];
const isPpiKaksTemporarilyUnavailable = stageId => stageId === 'ppi' || stageId === 'kaks';
const clearAutoExecuteTimer = () => { timerClears += 1; };
const getStageLabel = stageId => stageId === 'ppi' ? 'PPI' : stageId;
const ElMessage = {warning: message => warnings.push(message)};
"""
        + manual_block
        + """
doExecute().then(() => process.stdout.write(JSON.stringify({
  timerClears,
  warnings,
  dialogVisible: confirmDialogVisible.value,
  executing: executing.value,
  isExecuting: isExecuting.value
})));
"""
    )

    assert result == {
        "timerClears": 1,
        "warnings": ["PPI 功能修订中，当前暂不可用"],
        "dialogVisible": False,
        "executing": False,
        "isExecuting": False,
    }


def test_background_batch_applies_temporary_pauses_before_execution_loop() -> None:
    background = _source_block(
        _index_html(),
        "const startBackgroundBatchExecution = async",
        "const extractDataPathsFromText =",
    )

    pause_policy = background.index("ppiKaksTemporaryPauseStageIds")
    execution_loop = background.index("while (true)")
    direct_execute = background.index("executeStageDirect")
    assert pause_policy < execution_loop < direct_execute
    assert "localStatus[stageId] = 'paused'" in background


def test_background_batch_classifies_only_owner_preflight_blocks_as_paused() -> None:
    helper = _source_block(
        _index_html(),
        "const isPpiKaksBackgroundPause =",
        "const startBackgroundBatchExecution = async",
    )
    result = _run_node(
        helper
        + """
const cases = {
  kaks: isPpiKaksBackgroundPause('kaks', {}, {status: 'needs_reference_data'}, ['kaks']),
  ppi: isPpiKaksBackgroundPause('ppi', {}, {ppiSpeciesResolution: {status: 'confirmation_required'}}, ['ppi']),
  ownedOrtholog: isPpiKaksBackgroundPause(
    'ortholog', {autoInsertedFor: ['ppi']}, {status: 'needs_ortholog'}, ['ppi']
  ),
  explicitOrtholog: isPpiKaksBackgroundPause(
    'ortholog', {}, {status: 'needs_ortholog'}, ['ortholog', 'ppi']
  ),
  ordinary: isPpiKaksBackgroundPause('motif', {}, {status: 'needs_reference_data'}, ['motif'])
  ,temporaryPpi: isPpiKaksBackgroundPause('ppi', {}, {status: 'temporarily_unavailable'}, ['ppi'])
};
console.log(JSON.stringify(cases));
"""
    )

    assert result == {
        "kaks": True,
        "ppi": True,
        "ownedOrtholog": True,
        "explicitOrtholog": False,
        "ordinary": False,
        "temporaryPpi": True,
    }


def test_background_batch_reports_paused_separately_from_failed() -> None:
    background = _source_block(
        _index_html(),
        "const startBackgroundBatchExecution = async",
        "const extractDataPathsFromText =",
    )

    assert "localStatus[stageId] = 'paused'" in background
    assert "let paused = 0;" in background
    assert "暂停 ${paused}/${stageIds.length}" in background