add Entra device identity to computers and Azure/Thor import sources

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.

tests/test_csv_import.py

@@ -173,14 +173,17 @@ def test_generated_templates_only_contain_real_columns(db):
 
     A hand-written template set had invented columns on 7 of 11 tables and named
     a table that does not exist. Generating from the schema is what makes that
-    impossible - so assert the generator actually does it.
+    impossible - so assert the generator actually does it. Plugin-registered
+    virtual columns (machineassetnumber) are legitimate, not invented, so they
+    are allowed alongside the table's real columns.
     """
+    from shopdb.core.services.csvimport import virtual_columns_for
     for table in IMPORTABLE:
         text = generate_template(table)
         header = [line for line in text.splitlines() if line and not line.startswith('#')][0]
-        real = {c.name for c in importable_columns(table)}
+        allowed = {c.name for c in importable_columns(table)} | virtual_columns_for(table)
         for column in header.split(','):
-            assert column in real, '%s.csv offers a column that does not exist: %s' % (table, column)
+            assert column in allowed, '%s.csv offers a column that does not exist: %s' % (table, column)
 
 
 def test_every_template_includes_its_required_columns(db):
@@ -234,3 +237,46 @@ def test_a_dry_run_over_several_files_resolves_across_them(db):
     assert result.ok, [str(p) for p in result.problems]
     _db.session.rollback()
     assert count('assets') == 0
+
+
+def _example_files():
+    """Reconstruct each template's example rows as an importable CSV: the header
+    plus the live (uncommented) example data rows that follow it."""
+    files = {}
+    for table in IMPORTABLE:
+        header = None
+        rows = []
+        for line in generate_template(table).splitlines():
+            if not line.strip():
+                continue
+            if line.startswith('#'):
+                continue
+            if header is None:
+                header = line
+            else:
+                rows.append(line)
+        files[table] = '\n'.join([header] + rows) + '\n'
+    return files
+
+
+def test_template_examples_are_present_for_every_table(db):
+    """Each template ships at least one worked example row, so a downloaded
+    set is a usable starter workbook rather than a bare header."""
+    files = _example_files()
+    for table in IMPORTABLE:
+        rows = files[table].splitlines()[1:]
+        assert rows, '%s.csv has no example row' % table
+
+
+def test_template_example_set_imports_cleanly(db):
+    """The whole curated example set must import as-is: every name a later file
+    references is defined by an earlier one, with no self-references that the
+    validation pass cannot resolve. This is the promise the workbook makes."""
+    resolver = Resolver()
+    files = _example_files()
+    for table in IMPORTABLE:
+        result = import_csv(table, files[table], resolver=resolver)
+        assert result.ok, '%s.csv example failed: %s' % (
+            table, [str(p) for p in result.problems])
+    _db.session.rollback()
+

tests/test_import_sources.py

@@ -0,0 +1,201 @@
+"""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

tests/test_plugins/test_computers_import_machine_link.py

@@ -0,0 +1,187 @@
+"""computers.csv can carry the machine it drives, and the link is built on import.
+
+A site with a spreadsheet knows "this PC drives machine CNC-014" and wants to
+write that in the PC row, not hand-build an assetrelationships sheet keyed on
+ids it never sees. The importer accepts a virtual `machineassetnumber` column,
+keeps it out of the computers write, and hands it to the computers plugin after
+each PC row is written. The rules that matter for a cold user driving this off a
+spreadsheet:
+
+- a good number builds the PC-controls-machine link;
+- a number that has no machine, names this PC itself, names a location-only
+  marker, or is claimed twice in one file, is an AMBER warning and is skipped -
+  the PC still imports, the file is still ok;
+- a dry run builds nothing and, critically, sends no alert;
+- re-running the same file does not duplicate the link.
+"""
+import pytest
+
+from shopdb.core.services.csvimport import (
+    Resolver, import_csv, virtual_columns_for,
+)
+from shopdb.extensions import db as _db
+
+
+REFERENCE = [
+    ('assettypes', 'assettype\nComputer\nMachine\n'),
+    ('assetstatuses', 'status\nIn Use\n'),
+    ('locationtypes', 'locationtype\nBay\n'),
+]
+
+
+def _seed(resolver):
+    """Reference rows, a location, and the 'controls' relationship type."""
+    from shopdb.core.models.relationship import RelationshipType
+
+    for table, text in REFERENCE:
+        result = import_csv(table, text, resolver=resolver, commit=True)
+        assert result.ok, [str(p) for p in result.problems]
+    result = import_csv(
+        'locations', 'locationname,locationtypeid\nBay 3,Bay\n',
+        resolver=resolver, commit=True)
+    assert result.ok, [str(p) for p in result.problems]
+    if not RelationshipType.query.filter_by(relationshiptype='controls').first():
+        _db.session.add(RelationshipType(relationshiptype='controls'))
+    # import_csv writes into the session but never commits; commit the setup so
+    # a dry-run import's rollback only unwinds the dry run, not the fixtures.
+    _db.session.commit()
+
+
+def _assets(resolver, *rows):
+    """Create the given assets. Each row is (assetnumber, assettype)."""
+    body = 'assetnumber,assettypeid,statusid,locationid\n'
+    body += ''.join('%s,%s,In Use,Bay 3\n' % r for r in rows)
+    result = import_csv('assets', body, resolver=resolver, commit=True)
+    assert result.ok, [str(p) for p in result.problems]
+    _db.session.commit()
+
+
+def _import_computers(resolver, body, commit=True):
+    return import_csv('computers', body, resolver=resolver, commit=commit)
+
+
+def _controls_target(pc_assetnumber):
+    """Asset number of the machine this PC actively controls, or None."""
+    from shopdb.core.models import Asset
+    from shopdb.core.models.relationship import RelationshipType, AssetRelationship
+
+    pc = Asset.query.filter_by(assetnumber=pc_assetnumber).first()
+    controls = RelationshipType.query.filter_by(relationshiptype='controls').first()
+    rel = AssetRelationship.query.filter(
+        AssetRelationship.sourceassetid == pc.assetid,
+        AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
+        AssetRelationship.isactive.is_(True)).first()
+    if rel is None:
+        return None
+    return _db.session.get(Asset, rel.targetassetid).assetnumber
+
+
+def test_the_column_is_virtual_and_owned_by_the_computers_plugin(db):
+    """The plugin registers machineassetnumber; core stays ignorant of it."""
+    assert 'machineassetnumber' in virtual_columns_for('computers')
+
+
+def test_a_good_machine_number_builds_the_link(db):
+    resolver = Resolver()
+    _seed(resolver)
+    _assets(resolver, ('PC-001', 'Computer'), ('CNC-014', 'Machine'))
+    result = _import_computers(
+        resolver,
+        'assetid,hostname,machineassetnumber\nPC-001,WS-PC-001,CNC-014\n')
+    assert result.ok, [str(p) for p in result.problems]
+    assert result.warnings == []
+    assert _controls_target('PC-001') == 'CNC-014'
+
+
+def test_an_empty_machine_number_links_nothing_and_does_not_warn(db):
+    resolver = Resolver()
+    _seed(resolver)
+    _assets(resolver, ('PC-001', 'Computer'))
+    result = _import_computers(
+        resolver, 'assetid,hostname,machineassetnumber\nPC-001,WS-PC-001,\n')
+    assert result.ok
+    assert result.warnings == []
+    assert _controls_target('PC-001') is None
+
+
+def test_a_machine_number_with_no_machine_warns_and_skips(db):
+    """The PC still imports; the unresolved number is amber, not a failed file."""
+    resolver = Resolver()
+    _seed(resolver)
+    _assets(resolver, ('PC-001', 'Computer'))
+    result = _import_computers(
+        resolver, 'assetid,hostname,machineassetnumber\nPC-001,WS-PC-001,NOPE-99\n')
+    assert result.ok, 'an unresolved machine number must not fail the file'
+    assert len(result.warnings) == 1
+    warning = result.warnings[0]
+    assert warning.column == 'machineassetnumber'
+    assert warning.target == 'NOPE-99'
+    assert warning.action, 'a warning must carry the one corrective step'
+    assert _controls_target('PC-001') is None
+
+
+def test_a_pc_naming_its_own_asset_number_is_not_self_linked(db):
+    resolver = Resolver()
+    _seed(resolver)
+    _assets(resolver, ('PC-001', 'Computer'))
+    result = _import_computers(
+        resolver, 'assetid,hostname,machineassetnumber\nPC-001,WS-PC-001,PC-001\n')
+    assert result.ok
+    assert len(result.warnings) == 1
+    assert 'own asset number' in result.warnings[0].reason
+    assert _controls_target('PC-001') is None
+
+
+def test_two_rows_claiming_one_machine_link_only_the_first(db):
+    """The in-file fan-out guard: a copied cell must not steal the bay."""
+    resolver = Resolver()
+    _seed(resolver)
+    _assets(resolver,
+            ('PC-001', 'Computer'), ('PC-002', 'Computer'), ('CNC-014', 'Machine'))
+    result = _import_computers(
+        resolver,
+        'assetid,hostname,machineassetnumber\n'
+        'PC-001,WS-PC-001,CNC-014\n'
+        'PC-002,WS-PC-002,CNC-014\n')
+    assert result.ok
+    assert len(result.warnings) == 1
+    assert 'already claimed' in result.warnings[0].reason
+    assert _controls_target('PC-001') == 'CNC-014'
+    assert _controls_target('PC-002') is None
+
+
+def test_a_dry_run_builds_no_link_and_leaves_no_trace(db):
+    resolver = Resolver()
+    _seed(resolver)
+    _assets(resolver, ('PC-001', 'Computer'), ('CNC-014', 'Machine'))
+    result = _import_computers(
+        resolver,
+        'assetid,hostname,machineassetnumber\nPC-001,WS-PC-001,CNC-014\n',
+        commit=False)
+    assert result.ok
+    assert result.created == 1
+    _db.session.rollback()
+    assert _controls_target('PC-001') is None
+
+
+def test_reimporting_the_same_link_does_not_duplicate(db):
+    resolver = Resolver()
+    _seed(resolver)
+    _assets(resolver, ('PC-001', 'Computer'), ('CNC-014', 'Machine'))
+    body = 'assetid,hostname,machineassetnumber\nPC-001,WS-PC-001,CNC-014\n'
+    first = _import_computers(resolver, body)
+    assert first.ok and first.warnings == []
+
+    second = _import_computers(resolver, body)
+    assert second.ok and second.warnings == []
+    assert _controls_target('PC-001') == 'CNC-014'
+
+    from shopdb.core.models import Asset
+    from shopdb.core.models.relationship import RelationshipType, AssetRelationship
+    pc = Asset.query.filter_by(assetnumber='PC-001').first()
+    controls = RelationshipType.query.filter_by(relationshiptype='controls').first()
+    active = AssetRelationship.query.filter(
+        AssetRelationship.sourceassetid == pc.assetid,
+        AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
+        AssetRelationship.isactive.is_(True)).count()
+    assert active == 1, 'a re-import must not stack a second active link'