shopdb-flask/tests/test_csv_import.py

"""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.core.services.csvimport import digit_runs, normalize_label
from shopdb.extensions import db as _db


REFERENCE = [
    ('assettypes', 'assettype,description\nMachine,Shop floor\nPrinter,Office\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'
        'PRN-01,HP,Printer,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) == ('Printer', '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()


# -- fuzzy reconciliation safety (review item 1.1) -----------------------------
#
# A high similarity score is the DANGEROUS case: 'Bay 310' and 'Bay 311' share
# every word and score ~0.857, so a naive threshold silently merges two distinct
# locations. Reconciliation now ships OFF (threshold 0) and, even when a site
# turns it on, refuses any pair whose digit runs differ.

def test_digit_runs_extracts_number_groups():
    assert digit_runs(normalize_label('Bay 310')) == ['310']
    assert digit_runs(normalize_label('Precision 3660')) == ['3660']
    assert digit_runs(normalize_label('Dell Inc.')) == []


def test_reconciliation_is_off_by_default(db):
    """A fresh Resolver does not fuzzy-match: an unknown numbered bay is a row
    error, not a silent bind onto its neighbour."""
    resolver = Resolver()
    seed_reference(resolver)
    # 'Bay 3' exists from the reference seed; 'Bay 8' does not.
    result = import_csv(
        'assets',
        'assetnumber,assettypeid,locationid\nA-1,Machine,Bay 8\n',
        resolver=resolver)
    assert not result.ok
    assert not resolver.events


def test_reconcile_refuses_when_digit_runs_differ(db):
    """Even with fuzzy matching turned on, 'Bay 311' must not bind to the
    existing 'Bay 3': the digit runs differ, so it is a distinct location."""
    resolver = Resolver(reconcile_threshold=0.5)
    seed_reference(resolver)
    result = import_csv(
        'assets',
        'assetnumber,assettypeid,locationid\nB-1,Machine,Bay 311\n',
        resolver=resolver)
    assert not result.ok
    assert not resolver.events


def test_reconcile_still_binds_normalized_exact(db):
    """The normalized-exact tier is untouched: 'bay 3' folds to the existing
    'Bay 3' regardless of threshold, and is reported as an event."""
    resolver = Resolver(reconcile_threshold=0.9)
    seed_reference(resolver)
    result = import_csv(
        'assets',
        'assetnumber,assettypeid,locationid\nC-1,Machine,BAY 3\n',
        resolver=resolver)
    assert result.ok, [str(p) for p in result.problems]
    _db.session.rollback()




def test_a_numeric_assetnumber_is_read_as_a_name_not_an_id(db):
    """This site numbers operations 0NNN, so numeric natural keys are real.

    An all-digit string used to be taken at face value as an id. `0600` became
    id 600 - a different asset, or none - which is how an Azure device import
    died on a foreign key partway through, and how a PC could instead have been
    bound silently to a printer's record. The label wins when one matches.
    """
    resolver = Resolver()
    seed_reference(resolver)
    import_csv('assets', 'assetnumber,assettypeid\n0600,Machine\n',
               resolver=resolver, commit=True)
    realid = _db.session.execute(_db.text(
        "select assetid from assets where assetnumber = '0600'")).scalar()

    # Force the ambiguity: an id that is NOT the one named 0600.
    assert resolver.resolve('assets', 'assetid', '0600') == realid


def test_a_number_naming_nothing_is_still_read_as_an_id(db):
    """Round-tripped exports and sites that know their ids keep working."""
    resolver = Resolver()
    seed_reference(resolver)
    assert resolver.resolve('assets', 'assetid', '999999') == 999999