9ca562745a
feat(ticket): update email integration to use new priority constants for ticket classification feat(procurement): add procurement overview page with dynamic data loading and display test(subscriptions): add tests for billing calendar to ensure correct invoice dates feat(reminder): implement automated task lists with user-defined rules for reminders feat(migrations): create tables for managing delefiber product prices and mobile recorder provisioning history test(mobile_recorder): add tests for provisioning mobile recorders to ensure correct asset creation and updates
74 lines
2.8 KiB
Python
74 lines
2.8 KiB
Python
import asyncio
|
|
|
|
from app.modules.hardware.backend import router as hardware_router
|
|
|
|
|
|
def _payload(**overrides):
|
|
data = {
|
|
"name": "BMC Recorder #24",
|
|
"recorder_number": 24,
|
|
"serial_number": "SERIAL-24",
|
|
"udid": "udid-24",
|
|
"ecid": "ecid-24",
|
|
"imei": "imei-24",
|
|
"wifi_mac": "AA:BB:CC:DD:EE:FF",
|
|
"os_version": "18.0",
|
|
"supervised": True,
|
|
}
|
|
data.update(overrides)
|
|
return hardware_router.MobileRecorderProvisionRequest(**data)
|
|
|
|
|
|
def test_mobile_recorder_provisioning_creates_asset_by_serial(monkeypatch):
|
|
calls = []
|
|
|
|
def fake_query(query, params=None, fetch=True):
|
|
calls.append((query, params, fetch))
|
|
if "SELECT id, hardware_specs" in query:
|
|
return []
|
|
if "INSERT INTO hardware_assets" in query:
|
|
return [{"id": 1842}]
|
|
return []
|
|
|
|
monkeypatch.setattr(hardware_router, "execute_query", fake_query)
|
|
monkeypatch.setattr(hardware_router.settings, "MOBILE_RECORDER_PROVISIONING_TOKEN", "provision-token")
|
|
|
|
result = asyncio.run(hardware_router.provision_mobile_recorder(_payload(), "Bearer provision-token", None))
|
|
|
|
assert result == {"success": True, "action": "created", "asset_id": 1842, "recorder_number": 24}
|
|
insert = next(params for query, params, _ in calls if "INSERT INTO hardware_assets" in query)
|
|
assert insert[2] == "SERIAL-24"
|
|
assert insert[5].adapted["mobile_recorder"]["supervised"] is True
|
|
|
|
|
|
def test_mobile_recorder_provisioning_updates_existing_serial(monkeypatch):
|
|
calls = []
|
|
|
|
def fake_query(query, params=None, fetch=True):
|
|
calls.append((query, params, fetch))
|
|
if "SELECT id, hardware_specs" in query:
|
|
return [{"id": 99, "hardware_specs": {"other_connector": {"keep": True}}}]
|
|
if "UPDATE hardware_assets" in query:
|
|
return [{"id": 99}]
|
|
return []
|
|
|
|
monkeypatch.setattr(hardware_router, "execute_query", fake_query)
|
|
monkeypatch.setattr(hardware_router.settings, "MOBILE_RECORDER_PROVISIONING_TOKEN", "provision-token")
|
|
|
|
result = asyncio.run(hardware_router.provision_mobile_recorder(_payload(), None, "provision-token"))
|
|
|
|
assert result["action"] == "updated"
|
|
update = next(params for query, params, _ in calls if "UPDATE hardware_assets" in query)
|
|
assert update[-1] == 99
|
|
assert update[4].adapted["other_connector"] == {"keep": True}
|
|
assert update[4].adapted["mobile_recorder"]["udid"] == "udid-24"
|
|
|
|
|
|
def test_mobile_recorder_provisioning_requires_service_token(monkeypatch):
|
|
monkeypatch.setattr(hardware_router.settings, "MOBILE_RECORDER_PROVISIONING_TOKEN", "provision-token")
|
|
try:
|
|
asyncio.run(hardware_router.provision_mobile_recorder(_payload(), None, "wrong"))
|
|
assert False, "Expected an authentication error"
|
|
except hardware_router.HTTPException as exc:
|
|
assert exc.status_code == 401
|