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.
283 lines
11 KiB
Python
283 lines
11 KiB
Python
"""Bulk CSV import: name resolution, all-or-nothing safety, and template truth.
|
|
|
|
The behaviour under test is what makes a spreadsheet import usable by someone
|
|
who does not have the database in front of them. In particular: a foreign key
|
|
column accepts the NAME of the thing it points at. Requiring numeric ids means
|
|
importing a file, reading back generated ids, and pasting them into the next
|
|
file - a workflow nobody finishes.
|
|
"""
|
|
import pytest
|
|
|
|
from shopdb.core.services.csvimport import (
|
|
IMPORTABLE, ImportError_, Resolver, generate_template, import_csv,
|
|
importable_columns, required_columns,
|
|
)
|
|
from shopdb.extensions import db as _db
|
|
|
|
|
|
REFERENCE = [
|
|
('assettypes', 'assettype,description\nMachine,Shop floor\nMeasuring Tool,Gage lab\n'),
|
|
('assetstatuses', 'status,description\nActive,In service\nRetired,Out\n'),
|
|
('locationtypes', 'locationtype\nBay\nRoom\n'),
|
|
]
|
|
|
|
|
|
def seed_reference(resolver):
|
|
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,building,locationtypeid\nBay 3,Building 1,Bay\nGage Lab,Building 2,Room\n',
|
|
resolver=resolver, commit=True)
|
|
assert result.ok, [str(p) for p in result.problems]
|
|
|
|
|
|
def count(table):
|
|
return _db.session.execute(_db.text('select count(*) from %s' % table)).scalar()
|
|
|
|
|
|
def test_foreign_keys_resolve_by_name(db):
|
|
"""The headline behaviour: write 'Gage Lab', not locationid=2."""
|
|
resolver = Resolver()
|
|
seed_reference(resolver)
|
|
result = import_csv(
|
|
'assets',
|
|
'assetnumber,name,assettypeid,statusid,locationid\n'
|
|
'CMM-01,Zeiss,Measuring Tool,Active,Gage Lab\n',
|
|
resolver=resolver, commit=True)
|
|
assert result.ok, [str(p) for p in result.problems]
|
|
|
|
row = _db.session.execute(_db.text(
|
|
'select t.assettype, s.status, l.locationname from assets a '
|
|
'join assettypes t on t.assettypeid = a.assettypeid '
|
|
'join assetstatuses s on s.statusid = a.statusid '
|
|
'join locations l on l.locationid = a.locationid')).fetchone()
|
|
assert tuple(row) == ('Measuring Tool', 'Active', 'Gage Lab')
|
|
|
|
|
|
def test_numeric_ids_still_work(db):
|
|
"""Exports round-trip, and some sites genuinely know their ids."""
|
|
resolver = Resolver()
|
|
seed_reference(resolver)
|
|
typeid = _db.session.execute(_db.text(
|
|
"select assettypeid from assettypes where assettype = 'Machine'")).scalar()
|
|
result = import_csv(
|
|
'assets', 'assetnumber,assettypeid\nMILL-1,%d\n' % typeid,
|
|
resolver=resolver, commit=True)
|
|
assert result.ok, [str(p) for p in result.problems]
|
|
assert count('assets') == 1
|
|
|
|
|
|
def test_unknown_name_names_the_column_and_the_value(db):
|
|
"""The error has to be actionable by someone holding a spreadsheet."""
|
|
resolver = Resolver()
|
|
seed_reference(resolver)
|
|
result = import_csv(
|
|
'assets',
|
|
'assetnumber,assettypeid,locationid\nX-1,Machine,Bay 9\n',
|
|
resolver=resolver, commit=False)
|
|
assert not result.ok
|
|
message = str(result.problems[0])
|
|
assert "locationid" in message
|
|
assert "Bay 9" in message
|
|
assert "locations.csv" in message
|
|
|
|
|
|
def test_reimport_updates_rather_than_duplicating(db):
|
|
"""Sites correct their spreadsheet and run it again. That must not double up."""
|
|
resolver = Resolver()
|
|
seed_reference(resolver)
|
|
text = 'assetnumber,name,assettypeid\nCMM-01,Zeiss,Machine\n'
|
|
first = import_csv('assets', text, resolver=resolver, commit=True)
|
|
assert (first.created, first.updated) == (1, 0)
|
|
|
|
renamed = 'assetnumber,name,assettypeid\nCMM-01,Zeiss Contura,Machine\n'
|
|
second = import_csv('assets', renamed, resolver=resolver, commit=True)
|
|
assert (second.created, second.updated) == (0, 1)
|
|
assert count('assets') == 1
|
|
assert _db.session.execute(_db.text(
|
|
"select name from assets where assetnumber = 'CMM-01'")).scalar() == 'Zeiss Contura'
|
|
|
|
|
|
def test_one_bad_row_writes_nothing(db):
|
|
"""A typo on the last line must not leave the earlier lines imported."""
|
|
resolver = Resolver()
|
|
seed_reference(resolver)
|
|
result = import_csv(
|
|
'assets',
|
|
'assetnumber,assettypeid\n'
|
|
'GOOD-1,Machine\n'
|
|
'GOOD-2,Machine\n'
|
|
'BAD-3,NoSuchType\n',
|
|
resolver=resolver, commit=True)
|
|
assert not result.ok
|
|
assert count('assets') == 0, 'rows were written despite a failure'
|
|
|
|
|
|
def test_unknown_column_is_rejected_with_the_accepted_list(db):
|
|
"""The failure mode of a hand-written template: invented columns."""
|
|
resolver = Resolver()
|
|
seed_reference(resolver)
|
|
result = import_csv(
|
|
'vendors', 'vendor,zipcode\nDell,78682\n', resolver=resolver, commit=False)
|
|
assert not result.ok
|
|
message = str(result.problems[0])
|
|
assert 'zipcode' in message
|
|
assert 'this table accepts' in message
|
|
|
|
|
|
def test_missing_required_column_is_rejected(db):
|
|
result = import_csv('assets', 'name\nNo asset number\n', commit=False)
|
|
assert not result.ok
|
|
assert 'assetnumber' in str(result.problems[0])
|
|
|
|
|
|
def test_booleans_take_1_and_0_and_reject_prose(db):
|
|
resolver = Resolver()
|
|
seed_reference(resolver)
|
|
ok = import_csv('assets', 'assetnumber,assettypeid,isactive\nA-1,Machine,0\n',
|
|
resolver=resolver, commit=True)
|
|
assert ok.ok, [str(p) for p in ok.problems]
|
|
|
|
bad = import_csv('assets', 'assetnumber,assettypeid,isactive\nA-2,Machine,maybe\n',
|
|
resolver=resolver, commit=False)
|
|
assert not bad.ok
|
|
assert 'isactive' in str(bad.problems[0])
|
|
|
|
|
|
def test_comment_and_blank_lines_are_ignored(db):
|
|
"""The templates ship a commented example. Importing one unedited should be
|
|
a no-op, not a row of literal placeholder text."""
|
|
resolver = Resolver()
|
|
seed_reference(resolver)
|
|
result = import_csv(
|
|
'assets',
|
|
'# a comment\nassetnumber,assettypeid\n\n# <assetnumber>,<type>\nREAL-1,Machine\n',
|
|
resolver=resolver, commit=True)
|
|
assert result.ok, [str(p) for p in result.problems]
|
|
assert result.created == 1
|
|
|
|
|
|
def test_passwords_are_never_importable():
|
|
"""A CSV carrying password hashes gets mailed around; one carrying plaintext
|
|
is worse. Neither column may be set from a file."""
|
|
for table in IMPORTABLE:
|
|
names = {c.name for c in importable_columns(table)}
|
|
assert 'passwordhash' not in names
|
|
assert 'password' not in names
|
|
|
|
|
|
def test_generated_templates_only_contain_real_columns(db):
|
|
"""The regression that motivated all of this.
|
|
|
|
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. 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]
|
|
allowed = {c.name for c in importable_columns(table)} | virtual_columns_for(table)
|
|
for column in header.split(','):
|
|
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):
|
|
"""A template that omits a required column produces a file that cannot import."""
|
|
for table in IMPORTABLE:
|
|
header = [line for line in generate_template(table).splitlines()
|
|
if line and not line.startswith('#')][0].split(',')
|
|
for column in required_columns(table):
|
|
assert column in header, '%s.csv is missing required column %s' % (table, column)
|
|
|
|
|
|
def test_importable_tables_all_exist(db):
|
|
"""A registry entry naming a table that is not in the schema would fail only
|
|
when someone tried to use it."""
|
|
for table in IMPORTABLE:
|
|
assert table in _db.metadata.tables, '%s is registered but not in the schema' % table
|
|
|
|
|
|
def test_unimportable_table_is_refused(db):
|
|
with pytest.raises(ImportError_):
|
|
import_csv('auditlog', 'x\n1\n', commit=False)
|
|
|
|
|
|
def test_dry_run_reports_counts_then_rolls_back(db):
|
|
"""The number an operator needs before saying yes - and no trace afterwards.
|
|
|
|
Writes go into the transaction even on a dry run, which is what lets a later
|
|
file resolve names an earlier one created. The rollback is what makes it a
|
|
dry run.
|
|
"""
|
|
resolver = Resolver()
|
|
seed_reference(resolver)
|
|
result = import_csv(
|
|
'assets', 'assetnumber,assettypeid\nD-1,Machine\nD-2,Machine\n',
|
|
resolver=resolver)
|
|
assert result.ok
|
|
assert result.created == 2
|
|
_db.session.rollback()
|
|
assert count('assets') == 0
|
|
|
|
|
|
def test_a_dry_run_over_several_files_resolves_across_them(db):
|
|
"""The whole-folder check: assets must resolve locations that only exist
|
|
because an earlier file in the same run created them."""
|
|
resolver = Resolver()
|
|
seed_reference(resolver)
|
|
result = import_csv(
|
|
'assets',
|
|
'assetnumber,assettypeid,locationid\nX-1,Machine,Bay 3\n',
|
|
resolver=resolver)
|
|
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()
|
|
|