4a74a1a405
A freshly-imported DB left setup_complete unset, so the admin was bounced into the first-run wizard even though the instance is fully populated. The harness now sets setup_complete=true (it already mints the admin), so an imported instance goes straight to the app.
178 lines
7.1 KiB
Python
178 lines
7.1 KiB
Python
"""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._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 = []
|
|
|
|
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
|
|
(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()
|