bmc_hub/tests/test_case_commands.py

import asyncio
import importlib
from unittest.mock import MagicMock
import pytest
from fastapi import HTTPException

commands=importlib.import_module('app.modules.sag.backend.commands')
actor={'id':1,'username':'ct','is_superadmin':True}
case={'id':186,'titel':'Router','customer_id':8,'assigned_group_id':1,'deadline':None,'deferred_until':None}

def test_ambiguous_initials_rejected():
    cursor=MagicMock(); cursor.fetchall.return_value=[{'user_id':1,'username':'ct','full_name':'Christian Thomas'},{'user_id':2,'username':'other','full_name':'Carsten Thomsen'}]
    with pytest.raises(HTTPException): commands.resolve_user(cursor,'ct')

@pytest.mark.parametrize('action',[{'type':'estimate','minutes':True},{'type':'time','minutes':0,'description':'Test'},{'type':'note','scope':0,'description':'Test'},{'type':'unknown'}])
def test_invalid_actions_rejected(action):
    with pytest.raises(HTTPException): commands.normalize(MagicMock(),case,{'commands':[action]},actor)

def test_subcase_requires_creation_right():
    with pytest.raises(HTTPException) as error:
        commands.normalize(MagicMock(),case,{'commands':[{'type':'subcase','title':'Router'}]},{'id':1,'permissions':['cases.edit']})
    assert error.value.status_code==403

def test_child_scope_and_estimate_resolved():
    result=commands.normalize(MagicMock(),case,{'commands':[{'type':'subcase','title':'Router'},{'type':'estimate','scope':0,'minutes':30}]},actor)
    assert result[0]['child_index']==0
    assert result[1]['scope']==0

def test_defer_after_deadline_rejected():
    with pytest.raises(HTTPException):
        commands.normalize(MagicMock(),dict(case,deadline='2026-10-06'),{'commands':[{'type':'defer','date':'2026-10-07'}]},actor)

def test_comment_html_escaped():
    cursor=MagicMock(); cursor.fetchone.return_value={'id':7}
    assert commands.comment(cursor,186,'ct','<img src=x onerror=alert(1)>')==7
    assert '&lt;img' in cursor.execute.call_args.args[1][2]

def test_execute_rollback_on_failed_action(monkeypatch):
    conn=MagicMock(); cursor=conn.cursor.return_value.__enter__.return_value
    cursor.fetchone.side_effect=[case,None]
    monkeypatch.setattr(commands,'get_db_connection',lambda:conn)
    monkeypatch.setattr(commands,'release_db_connection',lambda value:None)
    monkeypatch.setattr(commands,'normalize',lambda *args:[{'type':'note','description':'Test'}])
    def explode(*args): raise HTTPException(400,detail='Failed')
    monkeypatch.setattr(commands,'apply',explode)
    with pytest.raises(HTTPException): asyncio.run(commands.run(186,{'commands':[],'request_id':'11111111-1111-1111-1111-111111111111'},actor,True))
    conn.rollback.assert_called(); conn.commit.assert_not_called()

def test_replay_does_not_execute(monkeypatch):
    import hashlib,json
    payload={'commands':[],'request_id':'11111111-1111-1111-1111-111111111111'}
    digest=hashlib.sha256(json.dumps({'comment':None,'commands':[],'internal':True},sort_keys=True,ensure_ascii=False).encode()).hexdigest()
    conn=MagicMock(); cursor=conn.cursor.return_value.__enter__.return_value
    cursor.fetchone.side_effect=[case,{'sag_id':186,'actor_id':1,'payload_hash':digest,'result':{'results':[{'id':9}]}}]
    monkeypatch.setattr(commands,'get_db_connection',lambda:conn)
    monkeypatch.setattr(commands,'release_db_connection',lambda value:None)
    apply=MagicMock(); monkeypatch.setattr(commands,'apply',apply)
    assert asyncio.run(commands.run(186,payload,actor,True))=={'results':[{'id':9}]}
    apply.assert_not_called()


def test_untrusted_derived_fields_are_removed():
    result=commands.normalize(MagicMock(),case,{'commands':[{'type':'todo','title':'Test','assigned_to_user_id':999,'user_id':999,'billing':{'method':'invoice'}}]},actor)
    assert 'assigned_to_user_id' not in result[0]
    assert 'user_id' not in result[0]
    assert 'billing' not in result[0]

def test_parser_field_aliases_and_subcase_scope():
    result=commands.normalize(MagicMock(),case,{'commands':[{'type':'subcase','title':'Test','scope':0},{'type':'note','text':'Note','scope':0}]},actor)
    assert result[0]['child_index']==0
    assert result[1]['description']=='Note'

def test_priority_uses_existing_urgent_value():
    result=commands.normalize(MagicMock(),case,{'commands':[{'type':'priority','priority':'kritisk','reason':'Internet nede'}]},actor)
    assert result[0]['priority']=='urgent'

def test_subcase_title_database_limit():
    with pytest.raises(HTTPException):
        commands.normalize(MagicMock(),case,{'commands':[{'type':'subcase','title':'x'*256,'scope':0}]},actor)



def test_command_options_are_read_only_and_release_connection(monkeypatch):
    conn=MagicMock(); cursor=conn.cursor.return_value.__enter__.return_value
    cursor.fetchone.return_value={'id':186}
    cursor.fetchall.side_effect=[[{'user_id':1,'username':'ct','full_name':'Christian Thomas'}],[{'id':2,'name':'Support'}],[{'id':187,'titel':'Router'}]]
    monkeypatch.setattr(commands,'get_db_connection',lambda:conn)
    release=MagicMock();monkeypatch.setattr(commands,'release_db_connection',release)
    result=asyncio.run(commands.command_options(186,actor))
    assert result['users'][0]['user_id']==1
    assert result['groups'][0]['name']=='Support'
    assert result['cases'][0]['id']==187
    assert all(call.args[0].startswith('SELECT') for call in cursor.execute.call_args_list)
    conn.rollback.assert_called_once();conn.commit.assert_not_called();release.assert_called_once_with(conn)