Release 0.8.0: Windows installer, twelve fixes from a real Server 2019 install, operator docs

tests/test_csv_import.py

@@ -0,0 +1,236 @@
+"""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.
+    """
+    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)}
+        for column in header.split(','):
+            assert column in real, '%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