bmc_hub/tests/test_workflow_api.py

from contextlib import contextmanager
from copy import deepcopy
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.workflows import router as workflow_api


DEFINITION = {'trigger': 'time.created', 'branches': [{'condition': {'field': 'description', 'operator': 'contains', 'value': 'Excel'}, 'actions': [{'type': 'set_billing', 'method': 'subscription'}]}], 'metadata': {'name': 'Excel', 'priority': 10, 'customer_id': '24'}}


@pytest.fixture
def api(monkeypatch):
    statements = []
    workflow = {'id': 1, 'name': 'Excel', 'customer_id': '24', 'priority': 10, 'enabled': False, 'active_version_id': None}
    version = {'id': 7, 'workflow_id': 1, 'version': 1, 'definition': deepcopy(DEFINITION)}
    class Cursor:
        def execute(self, sql, params=None):
            statements.append((sql, params))
            self.sql = sql
        def fetchone(self):
            if 'bmc_workflow_versions' in self.sql:
                return deepcopy(version)
            return deepcopy(workflow)
        def fetchall(self):
            return []
    @contextmanager
    def transaction():
        yield Cursor()
    monkeypatch.setattr(workflow_api.service, 'transaction', transaction)
    app = FastAPI()
    app.include_router(workflow_api.router, prefix='/api/v1')
    for dep in (workflow_api.view_access, workflow_api.manage_access, workflow_api.activate_access):
        app.dependency_overrides[dep] = lambda: {'id': 4}
    return TestClient(app), statements, version


def test_simulator_is_read_only_and_returns_trace(api):
    client, statements, _ = api
    result = client.post('/api/v1/workflows/1/simulate', json={'version_id': 7, 'context': {'customer_id': '24', 'description': 'Excel', 'agreements': [{'id': 3}]}})
    assert result.status_code == 200
    assert result.json()['result']['method'] == 'subscription'
    assert result.json()['result']['trace'][0]['matched'] is True
    assert all(sql.startswith('SELECT') for sql, _ in statements)
    assert result.json()['simulation_receipt']


def test_activation_requires_valid_simulation_receipt(api):
    client, statements, _ = api
    response = client.post('/api/v1/workflows/1/activate', json={'version_id': 7, 'simulation_receipt': 'forged'})
    assert response.status_code == 409
    assert all(sql.startswith('SELECT') for sql, _ in statements)


def test_tested_version_activation_never_updates_definition(api):
    client, statements, _ = api
    simulation = client.post('/api/v1/workflows/1/simulate', json={'version_id': 7, 'context': {}}).json()
    response = client.post('/api/v1/workflows/1/activate', json={'version_id': 7, 'simulation_receipt': simulation['simulation_receipt']})
    assert response.status_code == 200
    assert any(sql.startswith('INSERT INTO bmc_workflow_activation_history') for sql, _ in statements)
    assert not any(sql.startswith('UPDATE bmc_workflow_versions') for sql, _ in statements)


def test_receipt_bound_to_immutable_definition_and_actor(api):
    _, _, version = api
    token = workflow_api._receipt(1, version, 4)
    with pytest.raises(Exception) as changed_actor:
        workflow_api._verify_receipt(token, 1, version, 5)
    assert changed_actor.value.status_code == 409
    version['definition']['branches'][0]['actions'][0]['method'] = 'invoice'
    with pytest.raises(Exception) as changed_definition:
        workflow_api._verify_receipt(token, 1, version, 4)
    assert changed_definition.value.status_code == 409


def test_invalid_definition_rejected_before_writes(api):
    client, statements, _ = api
    result = client.post('/api/v1/workflows', json={'name': 'SQL', 'definition': {'trigger': 'time.created', 'branches': [{'condition': {'field': 'password', 'operator': 'eq', 'value': 'secret'}, 'actions': [{'type': 'set_billing', 'method': 'invoice'}]}]}})
    assert result.status_code == 422
    assert not statements


def test_unauthenticated_administration_rejected():
    app = FastAPI()
    app.include_router(workflow_api.router)
    client = TestClient(app)
    assert client.post('/workflows/1/disable').status_code == 401
    assert client.get('/workflows/catalog').status_code == 401


def test_simulator_requires_both_entity_reference_fields(api):
    client, statements, _ = api
    assert client.post('/api/v1/workflows/1/simulate', json={'entity_type': 'case'}).status_code == 422
    assert not statements


def test_receipt_expires(api, monkeypatch):
    _, _, version = api
    monkeypatch.setattr(workflow_api.time, 'time', lambda: 100)
    token = workflow_api._receipt(1, version, 4)
    monkeypatch.setattr(workflow_api.time, 'time', lambda: 3701)
    with pytest.raises(Exception) as exc:
        workflow_api._verify_receipt(token, 1, version, 4)
    assert exc.value.status_code == 409


def test_invalid_action_cannot_activate(api):
    client, statements, version = api
    version['definition']['branches'][0]['actions'] = [{'type': 'notify', 'message': 'Test'}]
    receipt = workflow_api._receipt(1, version, 4)
    response = client.post('/api/v1/workflows/1/activate', json={'version_id': 7, 'simulation_receipt': receipt})
    assert response.status_code == 422
    assert all(sql.startswith('SELECT') for sql, _ in statements)