bdef21da50
PC records now carry Entra/Azure device identity so imported machines keep a durable link back to their directory object instead of matching on hostname alone. Adds azuredeviceid, azureobjectid, and azuredevicecategory to the Computer model with matching migrations, surfaces them read-only in PCDetail as an Entra Device Info card, and lets PCForm edit the device id directly. The computers plugin registers a machineassetnumber import hook that reconciles PC-to-machine links during bulk import and warns on unresolved or conflicting relationships. Adds importsources adapters that normalize Azure device-group and Thor PC inventory exports into canonical ShopDB CSVs via hostname alignment, exposes register_import_hook and RowWarning on the shopdb.api contract surface (contract 0.21.0), and refreshes the Import Site Assets page with clearer source instructions. Includes reconcile_locations maintenance script and coverage for the new import paths.
202 lines
8.0 KiB
Python
202 lines
8.0 KiB
Python
"""Foreign import-source adapters: transforming Azure and Thor exports.
|
|
|
|
These adapters are pure - they turn a foreign CSV into the canonical per-table
|
|
CSVs the import engine already understands, and touch no database. The tests
|
|
check the transform output directly, so they run without app context.
|
|
"""
|
|
import csv
|
|
import io
|
|
|
|
import pytest
|
|
|
|
from shopdb.core.services.importsources import (
|
|
AzureDeviceGroupSource, ImportSourceError, ThorSource, detect_source,
|
|
get_source,
|
|
)
|
|
|
|
|
|
def _rows(text):
|
|
return list(csv.DictReader(io.StringIO(text)))
|
|
|
|
|
|
AZURE_CSV = (
|
|
'displayName,objectType,mail,userType,id,deviceId,userPrincipalName,isUser,isGroup,isGuest\n'
|
|
'F13HSGH4,device,,,11111111-1111-1111-1111-111111111111,aaaa1111-bbbb-2222-cccc-333333333333,,False,False,False\n'
|
|
'E3P5M0B4,device,,,22222222-2222-2222-2222-222222222222,dddd4444-eeee-5555-ffff-666666666666,,False,False,False\n'
|
|
)
|
|
|
|
|
|
def test_azure_matches_signature():
|
|
header = ['displayName', 'objectType', 'deviceId', 'id']
|
|
assert AzureDeviceGroupSource.matches(header)
|
|
assert detect_source(header) is AzureDeviceGroupSource
|
|
|
|
|
|
def test_azure_transform_builds_assets_and_computers():
|
|
produced = AzureDeviceGroupSource.transform(AZURE_CSV)
|
|
assert set(produced) == {'assets', 'computers'}
|
|
|
|
assets = _rows(produced['assets'])
|
|
assert [a['assetnumber'] for a in assets] == ['F13HSGH4', 'E3P5M0B4']
|
|
assert all(a['assettypeid'] == 'computer' for a in assets)
|
|
assert all(a['statusid'] == 'In Use' for a in assets)
|
|
|
|
computers = _rows(produced['computers'])
|
|
assert computers[0]['assetid'] == 'F13HSGH4'
|
|
assert computers[0]['hostname'] == 'F13HSGH4'
|
|
assert computers[0]['azuredeviceid'] == 'aaaa1111-bbbb-2222-cccc-333333333333'
|
|
assert 'computertypeid' not in computers[0]
|
|
|
|
|
|
def test_azure_transform_stamps_computertype():
|
|
produced = AzureDeviceGroupSource.transform(
|
|
AZURE_CSV, options={'computertype': 'Shopfloor PC'})
|
|
computers = _rows(produced['computers'])
|
|
assert all(c['computertypeid'] == 'Shopfloor PC' for c in computers)
|
|
|
|
|
|
def test_azure_transform_carries_objectid():
|
|
produced = AzureDeviceGroupSource.transform(AZURE_CSV)
|
|
computers = _rows(produced['computers'])
|
|
# The CSV 'id' column is the Entra directory object id.
|
|
assert computers[0]['azureobjectid'] == '11111111-1111-1111-1111-111111111111'
|
|
|
|
|
|
def test_azure_transform_stamps_category():
|
|
produced = AzureDeviceGroupSource.transform(
|
|
AZURE_CSV, options={'category': 'SFLD_DYN_Device_Group_X'})
|
|
computers = _rows(produced['computers'])
|
|
assert all(c['azuredevicecategory'] == 'SFLD_DYN_Device_Group_X'
|
|
for c in computers)
|
|
|
|
|
|
def test_azure_aligns_matched_hostname_to_existing_asset():
|
|
# A case-insensitive hostname match routes the device to its real asset and
|
|
# emits NO new asset row, so the import updates in place instead of creating
|
|
# a duplicate. Unmatched devices still create a new asset keyed on hostname.
|
|
produced = AzureDeviceGroupSource.transform(
|
|
AZURE_CSV, options={'hostname_map': {'f13hsgh4': '13NSGH4'}})
|
|
|
|
computers = _rows(produced['computers'])
|
|
by_host = {c['hostname']: c for c in computers}
|
|
# Matched device points at the existing assetnumber, not its hostname.
|
|
assert by_host['F13HSGH4']['assetid'] == '13NSGH4'
|
|
# Unmatched device falls back to its hostname (new asset).
|
|
assert by_host['E3P5M0B4']['assetid'] == 'E3P5M0B4'
|
|
|
|
# Only the unmatched device produces an asset row.
|
|
assets = _rows(produced['assets'])
|
|
assert [a['assetnumber'] for a in assets] == ['E3P5M0B4']
|
|
|
|
|
|
def test_azure_all_matched_emits_no_assets_file():
|
|
produced = AzureDeviceGroupSource.transform(
|
|
AZURE_CSV,
|
|
options={'hostname_map': {'f13hsgh4': '13NSGH4', 'e3p5m0b4': 'E9Q'}})
|
|
# Every device matched, so there is nothing to create and no assets file.
|
|
assert 'assets' not in produced
|
|
computers = _rows(produced['computers'])
|
|
assert {c['assetid'] for c in computers} == {'13NSGH4', 'E9Q'}
|
|
|
|
|
|
def test_azure_skips_non_device_rows_and_dedupes():
|
|
text = (
|
|
'displayName,objectType,mail,userType,id,deviceId,userPrincipalName,isUser,isGroup,isGuest\n'
|
|
'F13HSGH4,device,,,1,aaaa,,False,False,False\n'
|
|
'F13HSGH4,device,,,1,aaaa,,False,False,False\n'
|
|
'Some User,user,u@x,,2,,u@x,True,False,False\n'
|
|
)
|
|
computers = _rows(AzureDeviceGroupSource.transform(text)['computers'])
|
|
assert [c['hostname'] for c in computers] == ['F13HSGH4']
|
|
|
|
|
|
def test_azure_rejects_wrong_header():
|
|
with pytest.raises(ImportSourceError):
|
|
AzureDeviceGroupSource.transform('a,b,c\n1,2,3\n')
|
|
|
|
|
|
THOR_CSV = (
|
|
'"Host Name","IP Address","LOGON","OS Version","OS ID","Last Updated","Boot Time","Location","Model","BIOS Manf."\n'
|
|
'"F67HSGH4","10.0.0.5","F67HSGH4\\ShopFloor","10.0.26100","24H2","2026-01-02 03:04:05","2026-01-01 06:00:00","S13","Dell Pro Tower Plus QBT1250","Dell Inc."\n'
|
|
)
|
|
|
|
|
|
def test_thor_matches_signature():
|
|
header = ['Host Name', 'BIOS Manf.', 'OS Version', 'Model']
|
|
assert ThorSource.matches(header)
|
|
assert detect_source(header) is ThorSource
|
|
|
|
|
|
def test_thor_transform_builds_lookup_and_subtype_files():
|
|
produced = ThorSource.transform(THOR_CSV)
|
|
assert 'vendors' in produced
|
|
assert 'models' in produced
|
|
assert 'operatingsystems' in produced
|
|
assert 'assets' in produced
|
|
assert 'computers' in produced
|
|
|
|
assert _rows(produced['vendors'])[0]['vendor'] == 'Dell Inc.'
|
|
assert _rows(produced['models'])[0]['modelnumber'] == 'Dell Pro Tower Plus QBT1250'
|
|
assert _rows(produced['operatingsystems'])[0]['osname'] == '10.0.26100 (24H2)'
|
|
|
|
computer = _rows(produced['computers'])[0]
|
|
assert computer['hostname'] == 'F67HSGH4'
|
|
assert computer['vendorid'] == 'Dell Inc.'
|
|
assert computer['modelnumberid'] == 'Dell Pro Tower Plus QBT1250'
|
|
assert computer['osid'] == '10.0.26100 (24H2)'
|
|
assert computer['loggedinuser'] == 'ShopFloor'
|
|
assert computer['lastreporteddate'] == '2026-01-02 03:04:05'
|
|
assert computer['lastboottime'] == '2026-01-01 06:00:00'
|
|
|
|
asset = _rows(produced['assets'])[0]
|
|
assert asset['locationid'] == 'S13'
|
|
assert asset['assettypeid'] == 'computer'
|
|
assert asset['statusid'] == 'In Use'
|
|
|
|
|
|
THOR_TWO_CSV = (
|
|
'"Host Name","IP Address","LOGON","OS Version","OS ID","Last Updated","Boot Time","Location","Model","BIOS Manf."\n'
|
|
'"F67HSGH4","10.0.0.5","F67HSGH4\\ShopFloor","10.0.26100","24H2","2026-01-02 03:04:05","2026-01-01 06:00:00","S13","Dell Pro Tower Plus QBT1250","Dell Inc."\n'
|
|
'"FDKKSGH4","10.0.0.6","FDKKSGH4\\ShopFloor","10.0.26100","24H2","2026-01-03 03:04:05","2026-01-02 06:00:00","G8","Dell Pro Tower Plus QBT1250","Dell Inc."\n'
|
|
)
|
|
|
|
|
|
def test_thor_stamps_computertype():
|
|
computer = _rows(
|
|
ThorSource.transform(THOR_CSV, options={'computertype': 'Shop Floor PC'})['computers'])[0]
|
|
assert computer['computertypeid'] == 'Shop Floor PC'
|
|
|
|
|
|
def test_thor_aligns_matched_hostname_to_existing_asset():
|
|
# A case-insensitive hostname match routes the host to its real asset and
|
|
# emits NO new asset row, so the import updates in place instead of creating
|
|
# a duplicate. Unmatched hosts still create a new asset keyed on hostname.
|
|
produced = ThorSource.transform(
|
|
THOR_TWO_CSV, options={'hostname_map': {'f67hsgh4': 'PC-1001'}})
|
|
|
|
computers = _rows(produced['computers'])
|
|
by_host = {c['hostname']: c for c in computers}
|
|
assert by_host['F67HSGH4']['assetid'] == 'PC-1001'
|
|
assert by_host['FDKKSGH4']['assetid'] == 'FDKKSGH4'
|
|
|
|
assets = _rows(produced['assets'])
|
|
assert [a['assetnumber'] for a in assets] == ['FDKKSGH4']
|
|
|
|
|
|
def test_thor_all_matched_emits_no_assets_file():
|
|
produced = ThorSource.transform(
|
|
THOR_TWO_CSV,
|
|
options={'hostname_map': {'f67hsgh4': 'PC-1001', 'fdkksgh4': 'PC-1002'}})
|
|
assert 'assets' not in produced
|
|
computers = _rows(produced['computers'])
|
|
assert {c['assetid'] for c in computers} == {'PC-1001', 'PC-1002'}
|
|
|
|
|
|
def test_get_source_unknown_raises():
|
|
with pytest.raises(ImportSourceError):
|
|
get_source('nope')
|
|
|
|
|
|
def test_detect_source_none_for_unknown_header():
|
|
assert detect_source(['foo', 'bar']) is None
|