WJ legacy-import reference loader: harness + reference/employees stages

Layer 2 of the import design (see the memory + scratchpad/IMPORT-PLAN.md): a
SITE-SPECIFIC reference loader that maps WJ's classic-ASP schema onto the
maintained, schema-agnostic IMPORT-API contract. Other sites copy the pattern
against their own source DB; nobody runs this loader as-is.

Harness (scripts/site_imports/wjf/harness.py): builds the app against the
current DATABASE_URL (point it at a throwaway import DB), mints an unscoped
admin PAT in-process, and drives the real import endpoints through the app test
client with Authorization: Bearer + X-Import-Mode - exercising the same
routes/authz/validation an HTTP client would, no running server needed.
Read-only pymysql access to the three scratch source DBs; legacy-id -> new-id
crosswalks persist to JSON so a crashed run resumes and later stages resolve FKs.

Stages implemented + verified idempotent against a fresh scratch target
(shopdb_flask_import): reference (vendors 46, businessunits 13, operatingsystems
11) and employees (directory 415, re-run updated-not-duplicated). Remaining
stages (models, applications, assets hub + crosswalk, dependents, network, usb,
verify) are stubbed with the same shape; README documents the adoption playbook.

idmap.json is generated state (gitignored).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

scripts/site_imports/wjf/harness.py

@@ -0,0 +1,157 @@
+"""Import harness for the WJ classic-ASP -> flask reference loader.
+
+This is SITE-SPECIFIC reference glue, NOT product code. It maps West Jefferson's
+classic schema onto the maintained, schema-agnostic import contract in
+docs/IMPORT-API.md. Other sites copy this pattern against their own source DB;
+they do not run this loader.
+
+The harness builds the flask app against the current DATABASE_URL (point it at a
+throwaway import database), mints an unscoped admin PAT in-process, and drives
+the REAL import endpoints through the app test client with the X-Import-Mode
+header - so it exercises the same routes/authz/validation an HTTP client would,
+without needing a running server. Legacy-id -> new-id maps persist to JSON so a
+crashed run resumes and later stages resolve foreign keys.
+"""
+
+import json
+import os
+import pymysql
+
+from shopdb import create_app
+from shopdb.extensions import db
+
+
+def _source_password():
+    """Root password for the scratch source DBs: reuse the dev DATABASE_URL in
+    .env unless SOURCE_DB_PASSWORD overrides."""
+    if os.environ.get('SOURCE_DB_PASSWORD'):
+        return os.environ['SOURCE_DB_PASSWORD']
+    env = os.path.join(os.path.dirname(__file__), '..', '..', '..', '.env')
+    with open(os.path.abspath(env)) as handle:
+        for line in handle:
+            if line.startswith('DATABASE_URL='):
+                after = line.split('root:', 1)[1]
+                return after.split('@', 1)[0]
+    raise RuntimeError('could not resolve source DB password')
+
+
+class Source:
+    """Read-only access to the three scratch source databases."""
+
+    def __init__(self, host='127.0.0.1', port=3306, user='root', password=None):
+        self.password = password or _source_password()
+        self.host, self.port, self.user = host, port, user
+        self._conns = {}
+
+    def db(self, name):
+        if name not in self._conns:
+            self._conns[name] = pymysql.connect(
+                host=self.host, port=self.port, user=self.user,
+                password=self.password, database=name, charset='utf8mb4',
+                cursorclass=pymysql.cursors.DictCursor)
+        return self._conns[name]
+
+    def rows(self, dbname, sql, args=None):
+        with self.db(dbname).cursor() as cur:
+            cur.execute(sql, args or ())
+            return cur.fetchall()
+
+
+class IdMap:
+    """Persisted legacy-id -> new-id crosswalks, one namespace per entity."""
+
+    def __init__(self, path):
+        self.path = path
+        self._data = {}
+        if os.path.exists(path):
+            with open(path) as handle:
+                self._data = json.load(handle)
+
+    def put(self, namespace, legacy, new):
+        self._data.setdefault(namespace, {})[str(legacy)] = new
+        self.save()
+
+    def get(self, namespace, legacy, default=None):
+        return self._data.get(namespace, {}).get(str(legacy), default)
+
+    def has(self, namespace, legacy):
+        return str(legacy) in self._data.get(namespace, {})
+
+    def count(self, namespace):
+        return len(self._data.get(namespace, {}))
+
+    def save(self):
+        with open(self.path, 'w') as handle:
+            json.dump(self._data, handle, indent=1)
+
+
+class Harness:
+    """App + admin PAT + import-mode test client + id maps."""
+
+    def __init__(self, idmap_path=None):
+        self.app = create_app()
+        self.appcontext = self.app.app_context()
+        self.appcontext.push()
+        self._silence_sql_logging()
+        self.client = self.app.test_client()
+        self.secret = self._mint_admin_pat()
+        default = os.path.join(os.path.dirname(__file__), 'idmap.json')
+        self.ids = IdMap(idmap_path or default)
+        self.source = Source()
+        self.errors = []
+
+    def _silence_sql_logging(self):
+        import logging
+        logging.getLogger('sqlalchemy.engine').setLevel(logging.WARNING)
+
+    def _mint_admin_pat(self):
+        """Create (or reuse) an admin user and an unscoped admin PAT. Unscoped
+        because import mode + role gates require full owner authority
+        (docs/IMPORT-API.md). Returns the clear secret."""
+        from werkzeug.security import generate_password_hash
+        from shopdb.core.models import User, Role, ApiToken
+        admin = User.query.join(User.roles).filter(Role.rolename == 'admin').first()
+        if not admin:
+            role = Role.query.filter_by(rolename='admin').first() \
+                or Role(rolename='admin', description='Administrator')
+            db.session.add(role)
+            admin = User(username='importer', email='importer@localhost',
+                         passwordhash=generate_password_hash('x'), isactive=True)
+            admin.roles.append(role)
+            db.session.add(admin)
+            db.session.commit()
+        secret = ApiToken.generate_secret()
+        token = ApiToken(userid=admin.userid, name='wjf legacy import',
+                         tokenprefix=ApiToken.prefix_of(secret),
+                         tokenhash=ApiToken.hash_secret(secret), scopes=None)
+        db.session.add(token)
+        db.session.commit()
+        return secret
+
+    def _headers(self, import_mode):
+        # A PAT authenticates as Authorization: Bearer shopdb_pat_... (the
+        # before_request shim resolves it into a JWT identity). X-API-Key is for
+        # collector/service tokens, not personal tokens.
+        headers = {'Authorization': f'Bearer {self.secret}'}
+        if import_mode:
+            headers['X-Import-Mode'] = 'true'
+        return headers
+
+    def post(self, path, payload, import_mode=True):
+        """POST to an import endpoint. Returns (status, data). 409s are returned
+        (idempotency handled by the caller), other 4xx/5xx are recorded."""
+        resp = self.client.post(path, json=payload, headers=self._headers(import_mode))
+        body = resp.get_json() or {}
+        data = body.get('data', body)
+        if resp.status_code >= 400 and resp.status_code != 409:
+            self.errors.append((path, resp.status_code, payload, data))
+        return resp.status_code, data
+
+    def get(self, path):
+        resp = self.client.get(path, headers=self._headers(False))
+        body = resp.get_json() or {}
+        return resp.status_code, body.get('data', body)
+
+    def close(self):
+        self.ids.save()
+        self.appcontext.pop()