bmc_hub/tests/test_hardware_inventory.py

"""Gateway inventory data must survive reviewed PC/Mac asset creation."""
import asyncio
import json
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fastapi import HTTPException
from app.modules.hardware.backend import inventory


def test_windows_list_and_mac_detail_formats():
    raw = {'hardware': {'manufacturer':'Apple','model':'Mac mini','cpu':{'name':'M4'},'ram_gb':24}, 'os':{'caption':'macOS'}}
    for field, value in [('hardware_data',json.dumps(raw)), ('json_data',raw)]:
        item = inventory.inventory_item({'id':12,'computer_name':'MAC-12','serial_number':'SN12',field:value})
        assert (item['brand'],item['model'],item['cpu'],item['ram_gb'],item['os']) == ('Apple','Mac mini','M4',24,'macOS')
        assert item['hardware_specs']['gateway_inventory']['data'] == raw
        assert item['serial_number'] == 'SN12'


def response_session(status, payload):
    response = MagicMock(status=status)
    response.json = AsyncMock(return_value=payload)
    context = MagicMock()
    context.__aenter__ = AsyncMock(return_value=response)
    session = MagicMock()
    session.get.return_value = context
    client = MagicMock()
    client.__aenter__ = AsyncMock(return_value=session)
    return client, session


def test_search_uses_gateway_filter_and_pagination():
    client, session = response_session(200, {'data':[{'id':1,'computer_name':'PC'}],'pagination':{'total_pages':3}})
    with patch.object(inventory,'_apigw_headers',return_value={'Authorization':'Bearer fixture'}), patch.object(inventory.aiohttp,'ClientSession',return_value=client):
        result = asyncio.run(inventory.search_inventory('PC', 'name', 2))
    assert result['items'][0]['name'] == 'PC'
    assert result['pagination']['total_pages'] == 3
    assert session.get.call_args.kwargs['params'] == {'computer_name':'PC','per_page':20,'page':2}


def test_gateway_failure_is_sanitized_and_serial_not_found_is_empty():
    for status, by in [(403,'name'), (404,'serial')]:
        client, session = response_session(status, {'error':'private upstream data'})
        with patch.object(inventory,'_apigw_headers',return_value={}), patch.object(inventory.aiohttp,'ClientSession',return_value=client), patch.object(inventory,'find_inventory_in_list',AsyncMock(return_value=None)):
            if by == 'serial':
                assert asyncio.run(inventory.search_inventory('SN/12', by, 1))['items'] == []
                assert session.get.call_args.args[0].endswith('/serial/SN%2F12')
            else:
                with pytest.raises(HTTPException) as error:
                    asyncio.run(inventory.search_inventory('', by, 1))
                assert error.value.status_code == 502
                assert 'private' not in error.value.detail


def test_pair_fetches_trusted_gateway_record_and_merges_only_inventory():
    item = inventory.inventory_item({'id':12,'computer_name':'PC12','json_data':{'hardware':{'ram_gb':16}}})
    with patch.object(inventory,'fetch_inventory_record',AsyncMock(return_value=item)) as fetch, patch.object(inventory,'execute_query',side_effect=[[{'id':4}],[{'id':4}]]) as query:
        result = asyncio.run(inventory.pair_inventory(4,inventory.InventoryPairInput(inventory_id=12)))
    fetch.assert_awaited_once_with(12)
    sql, params = query.call_args.args
    assert "COALESCE(hardware_specs, '{}'::jsonb) ||" in sql
    assert params[0].adapted == item['hardware_specs']
    assert set(params[0].adapted) == {'gateway_inventory'}
    assert 'serial_number =' not in sql
    assert 'current_owner' not in sql
    assert result['hardware_id'] == 4


def test_missing_asset_and_gateway_failure_never_write():
    with patch.object(inventory,'execute_query',return_value=[]) as query, patch.object(inventory,'fetch_inventory_record',AsyncMock()) as fetch:
        with pytest.raises(HTTPException) as error:
            asyncio.run(inventory.pair_inventory(4,inventory.InventoryPairInput(inventory_id=12)))
        assert error.value.status_code == 404
        fetch.assert_not_awaited()
        assert query.call_count == 1
    with patch.object(inventory,'execute_query',return_value=[{'id':4}]) as query, patch.object(inventory,'fetch_inventory_record',AsyncMock(side_effect=HTTPException(502,'Unavailable'))):
        with pytest.raises(HTTPException):
            asyncio.run(inventory.pair_inventory(4,inventory.InventoryPairInput(inventory_id=12)))
        assert query.call_count == 1


def test_unpair_removes_only_inventory_key():
    with patch.object(inventory,'execute_query',return_value=[{'id':4}]) as query:
        assert asyncio.run(inventory.unpair_inventory(4))['unpaired'] is True
    assert "- 'gateway_inventory'" in query.call_args.args[0]
    assert query.call_args.args[1] == (4,)


def test_detail_fetch_normalizes_gateway_record():
    client, session = response_session(200, {'id':12,'computer_name':'Mac','json_data':{'os':{'caption':'macOS'}}})
    with patch.object(inventory,'_apigw_headers',return_value={}), patch.object(inventory.aiohttp,'ClientSession',return_value=client):
        result = asyncio.run(inventory.fetch_inventory_record(12))
    assert session.get.call_args.args[0].endswith('/api/inventory/12')
    assert result['os'] == 'macOS'


def test_pair_detail_404_falls_back_to_fresh_paginated_list():
    client, session = response_session(404, {})
    def context(status, payload):
        return response_session(status, payload)[0].__aenter__.return_value.get.return_value
    session.get.side_effect = [context(404, {}), context(200, {'data':[{'id':1}], 'pagination':{'total_pages':2}}), context(200, {'data':[{'id':'12','computer_name':'Mac','hardware_data':'{"hardware":{"ram_gb":24}}'}], 'pagination':{'total_pages':2}})]
    with patch.object(inventory,'_apigw_headers',return_value={}), patch.object(inventory.aiohttp,'ClientSession',return_value=client):
        item = asyncio.run(inventory.fetch_inventory_record(12))
    assert item['name'] == 'Mac'
    assert item['ram_gb'] == 24
    assert session.get.call_args.kwargs['params']['page'] == 2


def test_missing_inventory_and_list_failure_do_not_pair_wrong_machine():
    for status, expected in [(200,404),(503,502)]:
        client, session = response_session(404,{})
        list_client, _ = response_session(status,{'data':[{'id':99}], 'pagination':{'total_pages':1}})
        session.get.side_effect = [session.get.return_value,list_client.__aenter__.return_value.get.return_value]
        with patch.object(inventory,'_apigw_headers',return_value={}), patch.object(inventory.aiohttp,'ClientSession',return_value=client):
            with pytest.raises(HTTPException) as error:
                asyncio.run(inventory.fetch_inventory_record(12))
        assert error.value.status_code == expected