"""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# ,\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