Add the API import surface for legacy migrations (contract 0.8.0)
All checks were successful
CI / backend (push) Successful in 1m4s
CI / naming (push) Successful in 1s
CI / frontend (push) Successful in 7s

Goal: an LLM or script can migrate an entire legacy database using only
the HTTP API - original history preserved, safely re-runnable.

- X-Import-Mode header (admin only): create/update endpoints across 15
  timestamped entity types accept original createddate/modifieddate;
  helper exposed via shopdb.api (contract 0.7.0 -> 0.8.0).
- Exact-match natural-key lookup filters on 13 list endpoints for the
  lookup-then-upsert recipe.
- Selfhosted USB checkout/checkin accept backdated event times in
  import mode.
- docs/IMPORT-API.md: operator manual grounded in the real legacy
  schema - order of operations, full table-by-table mapping including
  the machines fan-out, idempotent Python importer with dry-run, parity
  checks, and decided dispositions for unmigrated tables (DNC config
  stays live-fed via the collector; supportteams/appowners map to the
  upcoming supportteams model).

635 tests pass; naming green; frontend untouched.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
cproudlock
2026-07-11 20:10:46 -04:00
parent f8e5109255
commit 46e50c07ff
24 changed files with 1006 additions and 11 deletions

View File

@@ -0,0 +1,209 @@
"""Import-mode behavior: admin-gated legacy timestamp + event backdating.
Covers deliverables of the import-API work:
- createddate/modifieddate passthrough on a reference entity (vendor) and an
asset plugin (computer), gated on admin + X-Import-Mode header.
- both negative cases: no header, and header sent by a non-admin.
- a natural-key exact-match lookup filter added for idempotent import.
- backdated USB checkout time in selfhosted mode.
"""
import pytest
from werkzeug.security import generate_password_hash
# Import the usb models at collection time so their tables register on the
# metadata before the per-test db fixture calls create_all. The usb plugin is
# disabled in the dev registry, so its tables would otherwise never be created.
from plugins.usb.models import USBDevice, USBCheckout # noqa: F401
IMPORT_HEADER = {'X-Import-Mode': 'true'}
LEGACY_CREATED = '2020-01-05 08:30:00'
LEGACY_MODIFIED = '2021-06-07T14:15:16'
def _headers(auth_headers, importmode=True):
merged = dict(auth_headers)
if importmode:
merged.update(IMPORT_HEADER)
return merged
@pytest.fixture
def asset_types(db):
"""Seed the asset types the plugin create endpoints look up by name."""
from shopdb.core.models import AssetType
for name in ('computer',):
db.session.add(AssetType(assettype=name, pluginname=name,
tablename=name, description=name))
db.session.commit()
@pytest.fixture
def creator_headers(client, db):
"""A NON-admin user granted only computers.create/edit, logged in.
Proves import mode requires admin: this user carries the write permission
but must NOT get timestamp passthrough.
"""
from shopdb.core.models import User, Role, Permission
perms = []
for name in ('computers.create', 'computers.edit'):
p = Permission(name=name, description=name, category='computers')
db.session.add(p)
perms.append(p)
role = Role(rolename='pcoperator', description='PC operator')
role.permissions.extend(perms)
db.session.add(role)
db.session.flush()
user = User(username='pcop', email='pcop@test.local',
passwordhash=generate_password_hash('testpass'))
user.roles.append(role)
db.session.add(user)
db.session.commit()
response = client.post('/api/auth/login',
json={'username': 'pcop', 'password': 'testpass'})
assert response.status_code == 200, response.get_json()
token = response.get_json()['data']['access_token']
return {'Authorization': f'Bearer {token}'}
# ---------------------------------------------------------------------------
# Timestamp passthrough (positive)
# ---------------------------------------------------------------------------
def test_admin_import_mode_preserves_vendor_timestamps(client, db, auth_headers):
"""Admin + header: vendor keeps the legacy createddate/modifieddate."""
payload = {'vendor': 'LegacyVendor',
'createddate': LEGACY_CREATED,
'modifieddate': LEGACY_MODIFIED}
response = client.post('/api/vendors', json=payload,
headers=_headers(auth_headers))
assert response.status_code == 201, response.get_json()
from shopdb.core.models import Vendor
v = Vendor.query.filter_by(vendor='LegacyVendor').first()
assert v.createddate.year == 2020 and v.createddate.month == 1
assert v.createddate.day == 5 and v.createddate.hour == 8
assert v.modifieddate.year == 2021 and v.modifieddate.month == 6
def test_admin_import_mode_preserves_computer_asset_timestamps(
client, db, auth_headers, asset_types):
"""Admin + header: the created asset row keeps its legacy createddate."""
payload = {'assetnumber': 'PC-IMPORT-1',
'hostname': 'legacy-pc',
'createddate': LEGACY_CREATED,
'modifieddate': LEGACY_MODIFIED}
response = client.post('/api/computers', json=payload,
headers=_headers(auth_headers))
assert response.status_code == 201, response.get_json()
from shopdb.core.models import Asset
asset = Asset.query.filter_by(assetnumber='PC-IMPORT-1').first()
assert asset.createddate.year == 2020
assert asset.modifieddate.year == 2021
# ---------------------------------------------------------------------------
# Timestamp passthrough (negative): ignored without header, or for non-admin
# ---------------------------------------------------------------------------
def test_no_header_ignores_timestamps(client, db, auth_headers):
"""Admin but NO X-Import-Mode header: createddate falls back to now."""
payload = {'vendor': 'NoHeaderVendor', 'createddate': LEGACY_CREATED}
response = client.post('/api/vendors', json=payload,
headers=_headers(auth_headers, importmode=False))
assert response.status_code == 201, response.get_json()
from shopdb.core.models import Vendor
v = Vendor.query.filter_by(vendor='NoHeaderVendor').first()
assert v.createddate.year >= 2024 # server-stamped now, not the 2020 payload
def test_non_admin_ignores_timestamps(client, db, creator_headers, asset_types):
"""Non-admin WITH the write permission + header: timestamps still ignored."""
payload = {'assetnumber': 'PC-NONADMIN-1',
'createddate': LEGACY_CREATED}
response = client.post('/api/computers', json=payload,
headers=_headers(creator_headers))
assert response.status_code == 201, response.get_json()
from shopdb.core.models import Asset
asset = Asset.query.filter_by(assetnumber='PC-NONADMIN-1').first()
assert asset.createddate.year >= 2024 # not backdated: caller is not admin
# ---------------------------------------------------------------------------
# Natural-key lookup filter (idempotency recipe support)
# ---------------------------------------------------------------------------
def test_vendor_exact_lookup_filter(client, db, auth_headers):
"""GET /api/vendors?vendor=<name> exact-matches one vendor for lookup."""
for name in ('Acme', 'AcmeTools', 'Beta'):
resp = client.post('/api/vendors', json={'vendor': name},
headers=auth_headers)
assert resp.status_code == 201, resp.get_json()
listed = client.get('/api/vendors?vendor=Acme', headers=auth_headers)
assert listed.status_code == 200
rows = listed.get_json()['data']
assert len(rows) == 1
assert rows[0]['vendor'] == 'Acme'
# ---------------------------------------------------------------------------
# Backdated USB checkout (selfhosted mode)
#
# The usb plugin is disabled in the dev registry, so its blueprint may not be
# registered. Exercise the selfhosted layer directly inside a request context
# carrying the admin JWT + import header, which is what the route delegates to.
# ---------------------------------------------------------------------------
def _seed_usb_device(db, serialnumber):
from plugins.usb.models import USBDevice
device = USBDevice(serialnumber=serialnumber, ischeckedout=False, isactive=True)
db.session.add(device)
db.session.commit()
return device
def test_usb_backdated_checkout(app, db, auth_headers):
"""Admin + header: a USB checkout records the historical checkouttime."""
from plugins.usb.api import selfhosted
from plugins.usb.models import USBDevice, USBCheckout
device = _seed_usb_device(db, 'USB-IMPORT-1')
headers = _headers(auth_headers)
with app.test_request_context('/api/usb/USB-IMPORT-1/checkout',
method='POST', headers=headers):
resp = selfhosted.checkout_device(
'USB-IMPORT-1',
{'badge': '570005354', 'checkouttime': '2020-03-04 09:10:11'})
assert resp.status_code == 201, resp.get_json()
row = USBCheckout.query.filter_by(usbdeviceid=device.usbdeviceid).first()
assert row.checkouttime.year == 2020 and row.checkouttime.month == 3
reloaded = db.session.get(USBDevice, device.usbdeviceid)
assert reloaded.currentcheckoutdate.year == 2020
def test_usb_checkout_ignores_backdate_without_header(app, db, auth_headers):
"""No import header: the checkouttime override is ignored, uses now."""
from plugins.usb.api import selfhosted
from plugins.usb.models import USBDevice, USBCheckout
device = _seed_usb_device(db, 'USB-IMPORT-2')
with app.test_request_context('/api/usb/USB-IMPORT-2/checkout',
method='POST', headers=auth_headers):
resp = selfhosted.checkout_device(
'USB-IMPORT-2',
{'badge': '570005354', 'checkouttime': '2020-03-04 09:10:11'})
assert resp.status_code == 201, resp.get_json()
row = USBCheckout.query.filter_by(usbdeviceid=device.usbdeviceid).first()
assert row.checkouttime.year >= 2024 # server now, not the 2020 override