poyee-data-warehouse/tests/unit/ds/test_workflow.py

# -*- coding:utf-8 -*-
from unittest.mock import MagicMock

import pytest

from dw_base.ds.workflow import (
    find_schedule_by_workflow,
    find_workflow_by_name,
    gen_task_codes,
    get_workflow_full,
    list_workflow_names,
    release_schedule,
    release_workflow,
)


PC = 999  # 项目码占位


def _make_client():
    return MagicMock()


# --- find_workflow_by_name ---

def test_find_workflow_by_name_hit():
    client = _make_client()
    client.get.return_value = {
        'code': 0,
        'data': {'totalList': [
            {'name': 'wf_a', 'code': 100},
            {'name': 'wf_b', 'code': 200},
        ]}
    }
    result = find_workflow_by_name(client, PC, 'wf_b')
    assert result == {'name': 'wf_b', 'code': 200}
    client.get.assert_called_once_with(
        '/projects/999/workflow-definition',
        params={'pageNo': 1, 'pageSize': 200, 'searchVal': ''},
        expect_business_ok=True,
    )


def test_find_workflow_by_name_miss_returns_none():
    client = _make_client()
    client.get.return_value = {'code': 0, 'data': {'totalList': []}}
    assert find_workflow_by_name(client, PC, 'wf_x') is None


def test_find_workflow_by_name_custom_page_size():
    client = _make_client()
    client.get.return_value = {'code': 0, 'data': {'totalList': []}}
    find_workflow_by_name(client, PC, 'x', page_size=50)
    assert client.get.call_args[1]['params']['pageSize'] == 50


# --- list_workflow_names ---

def test_list_workflow_names():
    client = _make_client()
    client.get.return_value = {
        'code': 0,
        'data': {'totalList': [
            {'name': 'a'}, {'name': 'b'}, {'name': 'c'},
        ]}
    }
    assert list_workflow_names(client, PC) == ['a', 'b', 'c']


def test_list_workflow_names_empty():
    client = _make_client()
    client.get.return_value = {'code': 0, 'data': {'totalList': []}}
    assert list_workflow_names(client, PC) == []


# --- get_workflow_full ---

def test_get_workflow_full():
    client = _make_client()
    payload = {
        'workflowDefinition': {'code': 100, 'name': 'x'},
        'taskDefinitionList': [{'code': 1, 'version': 1}],
        'workflowTaskRelationList': [],
    }
    client.get.return_value = {'code': 0, 'data': payload}
    assert get_workflow_full(client, PC, 100) == payload
    client.get.assert_called_once_with(
        '/projects/999/workflow-definition/100',
        expect_business_ok=True,
    )


# --- gen_task_codes ---

def test_gen_task_codes_default_one():
    client = _make_client()
    client.get.return_value = {'code': 0, 'data': [171234567890]}
    codes = gen_task_codes(client, PC)
    assert codes == [171234567890]
    client.get.assert_called_once_with(
        '/projects/999/task-definition/gen-task-codes',
        params={'genNum': 1},
        expect_business_ok=True,
    )


def test_gen_task_codes_multi():
    client = _make_client()
    client.get.return_value = {'code': 0, 'data': [1, 2, 3]}
    assert gen_task_codes(client, PC, n=3) == [1, 2, 3]
    assert client.get.call_args[1]['params']['genNum'] == 3


# --- find_schedule_by_workflow ---

def test_find_schedule_hit():
    client = _make_client()
    client.get.return_value = {
        'code': 0,
        'data': {'totalList': [
            {'id': 5, 'workflowDefinitionCode': 100, 'releaseState': 'ONLINE'},
        ]}
    }
    s = find_schedule_by_workflow(client, PC, 100)
    assert s['id'] == 5
    assert s['releaseState'] == 'ONLINE'
    client.get.assert_called_once_with(
        '/projects/999/schedules',
        params={'pageNo': 1, 'pageSize': 100, 'searchVal': '',
                'workflowDefinitionCode': 100},
        expect_business_ok=True,
    )


def test_find_schedule_miss_returns_none():
    client = _make_client()
    client.get.return_value = {'code': 0, 'data': {'totalList': []}}
    assert find_schedule_by_workflow(client, PC, 100) is None


def test_find_schedule_filters_mismatched_wf_code():
    """totalList 可能含其他 workflow 的 schedule（保险二次过滤）。"""
    client = _make_client()
    client.get.return_value = {
        'code': 0,
        'data': {'totalList': [
            {'id': 1, 'workflowDefinitionCode': 999},
            {'id': 2, 'workflowDefinitionCode': 100},
        ]}
    }
    s = find_schedule_by_workflow(client, PC, 100)
    assert s['id'] == 2


# --- release_workflow ---

def test_release_workflow_online():
    client = _make_client()
    client.post.return_value = {'code': 0}
    release_workflow(client, PC, 100, 'ONLINE')
    client.post.assert_called_once_with(
        '/projects/999/workflow-definition/100/release',
        form_data={'releaseState': 'ONLINE'},
        expect_business_ok=True,
    )


def test_release_workflow_offline():
    client = _make_client()
    client.post.return_value = {'code': 0}
    release_workflow(client, PC, 100, 'OFFLINE')
    assert client.post.call_args[1]['form_data'] == {'releaseState': 'OFFLINE'}


def test_release_workflow_invalid_state_raises():
    client = _make_client()
    with pytest.raises(ValueError, match="ONLINE.*OFFLINE"):
        release_workflow(client, PC, 100, 'PAUSE')
    client.post.assert_not_called()


# --- release_schedule ---

def test_release_schedule_online_uses_online_endpoint():
    client = _make_client()
    client.post.return_value = {'code': 0}
    release_schedule(client, PC, 5, 'ONLINE')
    client.post.assert_called_once_with(
        '/projects/999/schedules/5/online',
        form_data={},
        expect_business_ok=True,
    )


def test_release_schedule_offline_uses_offline_endpoint():
    client = _make_client()
    client.post.return_value = {'code': 0}
    release_schedule(client, PC, 5, 'OFFLINE')
    assert client.post.call_args[0][0] == '/projects/999/schedules/5/offline'


def test_release_schedule_invalid_state_raises():
    client = _make_client()
    with pytest.raises(ValueError, match="ONLINE.*OFFLINE"):
        release_schedule(client, PC, 5, 'PAUSE')
    client.post.assert_not_called()