shopdb-flask/scripts/site_imports/wjf/harness.py

"""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
the IMPORT-API wiki page. 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._ensure_asset_types()
        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 = []
        self._defaultlevelid = None

    @property
    def defaultlevelid(self):
        """The level imported map positions belong to (ADR-017).

        The legacy schema predates levels: it has ONE floor plan, so every
        mapleft/maptop it carries is a coordinate on this site's default level.
        Importing them without a level produces markers the map refuses to draw
        - it will not guess a drawing for coordinates that do not name one.
        """
        if self._defaultlevelid is None:
            from shopdb.core.models import MapLevel
            level = MapLevel.default_level()
            self._defaultlevelid = level.levelid if level else None
        return self._defaultlevelid

    def _silence_sql_logging(self):
        import logging
        logging.getLogger('sqlalchemy.engine').setLevel(logging.WARNING)

    def _ensure_asset_types(self):
        """Register each plugin's AssetType + seeded subtypes. on_install (which
        does this) is skipped when a DB is built with `flask plugin upgrade-all`
        rather than a fresh `install`, so run it idempotently here - the create
        routes 500 without the AssetType row."""
        manager = self.app.extensions.get('plugin_manager')
        if not manager:
            return
        for _name, plugin in manager.get_all_plugins().items():
            try:
                plugin.on_install(self.app)
            except Exception:
                pass
        # A full import IS the setup, so mark it complete - otherwise the admin
        # is bounced to the first-run wizard on every fresh import DB.
        from shopdb.core.models import Setting
        Setting.set('setup_complete', 'true', valuetype='boolean', category='site')
        db.session.commit()

    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
        (the IMPORT-API wiki page). 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()