91 lines
3.9 KiB
Python
91 lines
3.9 KiB
Python
import pytest
|
|
from hardware_fakes import FakeAcceleratorCollector, FakeHostCollector, accelerator
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from modelforge_api.domain.enums import Availability, HardwareStatus
|
|
from modelforge_api.domain.hardware import HardwareSnapshot
|
|
from modelforge_api.persistence.models import (
|
|
Accelerator,
|
|
AuditEvent,
|
|
Base,
|
|
ComputeNode,
|
|
HardwareInventoryRun,
|
|
)
|
|
from modelforge_api.services.hardware_inventory import HardwareInventoryService
|
|
|
|
|
|
def service(session: Session, devices=None, availability=Availability.KNOWN, utilization=25):
|
|
return HardwareInventoryService(
|
|
session, FakeHostCollector(), FakeAcceleratorCollector(devices, availability, utilization)
|
|
)
|
|
|
|
|
|
def test_initial_creation_repeat_rediscovery_missing_and_return() -> None:
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
first = service(session, [accelerator()]).refresh()
|
|
assert first.overview.node_count == 1 and first.overview.accelerator_count == 1
|
|
service(session, [accelerator()]).refresh()
|
|
assert len(list(session.scalars(select(ComputeNode)))) == 1
|
|
assert len(list(session.scalars(select(Accelerator)))) == 1
|
|
missing = service(session, []).refresh().nodes[0].accelerators[0]
|
|
assert missing.status is HardwareStatus.MISSING
|
|
returned = service(session, [accelerator()]).refresh().nodes[0].accelerators[0]
|
|
assert returned.status is HardwareStatus.ACTIVE
|
|
assert len(list(session.scalars(select(Accelerator)))) == 1
|
|
actions = [event.action for event in session.scalars(select(AuditEvent))]
|
|
assert actions.count("ACCELERATOR_DISCOVERED") == 1
|
|
assert "ACCELERATOR_MISSING" in actions
|
|
|
|
|
|
def test_failed_enumeration_does_not_mark_known_gpu_missing() -> None:
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
service(session, [accelerator()]).refresh()
|
|
state = service(session, [], Availability.UNAVAILABLE).refresh()
|
|
assert state.nodes[0].accelerators[0].status is HardwareStatus.ACTIVE
|
|
assert state.overview.inventory_state is HardwareStatus.DEGRADED
|
|
|
|
|
|
def test_collector_failure_is_persisted_and_audited() -> None:
|
|
class FailingHostCollector:
|
|
def collect(self):
|
|
raise RuntimeError("host probe failed")
|
|
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
inventory = HardwareInventoryService(
|
|
session, FailingHostCollector(), FakeAcceleratorCollector([])
|
|
)
|
|
with pytest.raises(RuntimeError, match="host probe failed"):
|
|
inventory.refresh()
|
|
run = session.scalar(select(HardwareInventoryRun))
|
|
event = session.scalar(select(AuditEvent))
|
|
assert run is not None and run.status == "failed"
|
|
assert run.error == "RuntimeError: host probe failed"
|
|
assert event is not None and event.action == "INVENTORY_FAILED"
|
|
|
|
|
|
def test_fingerprint_determinism_ignores_telemetry_and_changes_for_driver() -> None:
|
|
host = FakeHostCollector().collect()
|
|
first = HardwareSnapshot(
|
|
host=host, nvidia=FakeAcceleratorCollector([accelerator()], utilization=1).collect()
|
|
)
|
|
telemetry_changed = HardwareSnapshot(
|
|
host=host, nvidia=FakeAcceleratorCollector([accelerator()], utilization=99).collect()
|
|
)
|
|
driver_changed_device = accelerator().model_copy(
|
|
update={
|
|
"driver_version": accelerator().driver_version.model_copy(update={"value": "601.0"})
|
|
}
|
|
)
|
|
driver_changed = HardwareSnapshot(
|
|
host=host, nvidia=FakeAcceleratorCollector([driver_changed_device]).collect()
|
|
)
|
|
assert first.fingerprint == telemetry_changed.fingerprint
|
|
assert first.fingerprint != driver_changed.fingerprint
|