80 lines
4.3 KiB
Python
80 lines
4.3 KiB
Python
"""Asset usage is independent of hardware type, ownership and rental lifecycle."""
|
|
import asyncio
|
|
import os
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from uuid import uuid4
|
|
|
|
from fastapi import HTTPException
|
|
from app.modules.hardware.backend import router as hardware
|
|
|
|
|
|
class AssetCategoryTests(unittest.TestCase):
|
|
def test_only_bmc_assets_can_be_classified(self):
|
|
for category in ('rental', 'internal'):
|
|
hardware._validate_bmc_asset_category(category, 'bmc')
|
|
with self.assertRaises(HTTPException):
|
|
hardware._validate_bmc_asset_category(category, 'customer')
|
|
hardware._validate_bmc_asset_category(None, 'customer')
|
|
hardware._validate_bmc_asset_category(None, 'bmc')
|
|
with self.assertRaises(HTTPException):
|
|
hardware._validate_bmc_asset_category('pc', 'bmc')
|
|
|
|
def test_existing_create_payload_without_category_still_works(self):
|
|
calls = []
|
|
def query(sql, params):
|
|
calls.append((sql, params))
|
|
self.assertEqual(sql.count('%s'), len(params))
|
|
return [{'id': 1, 'brand': None, 'model': 'Legacy PC'}]
|
|
with patch.object(hardware, 'execute_query', query):
|
|
result = asyncio.run(hardware.create_hardware({'asset_type':'pc', 'model':'Legacy PC'}))
|
|
self.assertEqual(result['id'], 1)
|
|
self.assertIsNone(calls[0][1][-1])
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get('BMC_HARDWARE_CATEGORY_DB_TESTS') == '1', 'Opt-in rollback-only DB test')
|
|
class AssetCategoryDatabaseTests(unittest.TestCase):
|
|
def test_create_edit_filter_and_customer_separation(self):
|
|
import psycopg2
|
|
from psycopg2.extras import RealDictCursor
|
|
from app.core.config import settings
|
|
from app.modules.rentals.backend import router as rentals
|
|
conn = psycopg2.connect(settings.DATABASE_URL)
|
|
def query(sql, params=None):
|
|
with conn.cursor(cursor_factory=RealDictCursor) as cursor:
|
|
cursor.execute(sql, params)
|
|
return list(cursor.fetchall()) if cursor.description else cursor.rowcount
|
|
token = uuid4().hex
|
|
try:
|
|
with patch.object(hardware, 'execute_query', query), patch.object(rentals, 'execute_query', query):
|
|
created = asyncio.run(hardware.create_hardware(dict(asset_type='pc',model='Asset test '+token,
|
|
current_owner_type='bmc',bmc_asset_category='rental')))
|
|
asset_id = created['id']
|
|
self.assertEqual((created['asset_type'],created['bmc_asset_category']), ('pc','rental'))
|
|
def listed(category):
|
|
return asyncio.run(rentals.list_assets(status='all', customer_id=None,
|
|
only_rental_enabled=False, bmc_asset_category=category))
|
|
self.assertIn(asset_id, [row['id'] for row in listed('rental')])
|
|
edited = asyncio.run(hardware.update_hardware(asset_id, {'bmc_asset_category':'internal'}))
|
|
self.assertEqual((edited['asset_type'],edited['current_owner_type'],edited['bmc_asset_category']), ('pc','bmc','internal'))
|
|
self.assertNotIn(asset_id, [row['id'] for row in listed('rental')])
|
|
self.assertIn(asset_id, [row['id'] for row in listed('internal')])
|
|
with self.assertRaises(HTTPException):
|
|
asyncio.run(hardware.update_hardware(asset_id, {'bmc_asset_category':'other'}))
|
|
cleared = asyncio.run(hardware.update_hardware(asset_id, {'bmc_asset_category':None}))
|
|
self.assertIsNone(cleared['bmc_asset_category'])
|
|
self.assertIn(asset_id, [row['id'] for row in listed('unassigned')])
|
|
customer = query("SELECT id FROM customers WHERE is_active=TRUE LIMIT 1")[0]['id']
|
|
customer_pc = asyncio.run(hardware.create_hardware(dict(asset_type='pc',model='Customer test '+token,
|
|
current_owner_type='customer',current_owner_customer_id=customer)))
|
|
self.assertNotIn(customer_pc['id'], [row['id'] for row in listed(None)])
|
|
with self.assertRaises(HTTPException):
|
|
asyncio.run(hardware.update_hardware(customer_pc['id'], {'bmc_asset_category':'internal'}))
|
|
finally:
|
|
conn.rollback()
|
|
conn.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|