diff --git a/scripts/site_imports/wjf/PILOT-DEPLOY.md b/scripts/site_imports/wjf/PILOT-DEPLOY.md index 7bd2a0e6..119ccba8 100644 --- a/scripts/site_imports/wjf/PILOT-DEPLOY.md +++ b/scripts/site_imports/wjf/PILOT-DEPLOY.md @@ -78,9 +78,36 @@ import API. It is site glue, not product code. venv/bin/python -m scripts.site_imports.wjf.run ``` - The 16 stages run in order (reference -> catalog -> assets hub -> locations -> - printers -> dependents -> relationships -> subnets -> usb -> verify). It is - idempotent - a crashed run resumes from `idmap.json`. + The 17 stages run in order (reference -> employees -> catalog -> assets hub + -> locations -> printers -> metrology -> communications -> applications -> + warranties -> notifications -> knowledgebase -> relationships -> + defaultprinters -> subnets -> usb -> verify). + + **Pointing it at a different source.** The three source databases are + named logically in the loader and resolved from the environment, so a + restore under another name needs no code edit: + + ``` + SOURCE_DB_MAIN=prodscratch \ + SOURCE_DB_USB=cmmc_usb_src \ + SOURCE_DB_EMPLOYEES=wjf_employees_src \ + DATABASE_URL=... venv/bin/python -m scripts.site_imports.wjf.run + ``` + + Also `SOURCE_DB_HOST`, `SOURCE_DB_PORT`, `SOURCE_DB_USER`, + `SOURCE_DB_PASSWORD`. The run prints every source it redirected, so + "it imported nothing" can be told apart from "it read the wrong database". + + **Resume is per target database.** A crashed run resumes from the crosswalk + in `idmap.json`, which maps this site's legacy ids to assetids **in one + specific database**. The loader stamps that database's identity into the + file and refuses a crosswalk built against a different one, because reusing + it attaches warranties and printer assignments to whatever asset now holds + that id - wrong data, no error. If you rebuild the target, delete + `idmap.json` or pass `--idmap` a fresh path. + + The exit code means something: non-zero if any endpoint errored, so a run + that logged 200 failures cannot read as a success in a deploy log. 3. Reclassify servers into network devices. The classic DB stored servers as PCs, so the import lands them as `computer` assets. Re-point them in place: @@ -151,8 +178,12 @@ documented skips (inactive rows, duplicate machinenumbers, LocationOnly, the - Keep the classic app authoritative during the window. The pilot is read-mostly for validation; do not dual-write. - Have a few real users (IT + a floor lead) work the pilot and log gaps. -- Re-import is cheap: fix a loader mapping, drop + rebuild the pilot DB, re-run. - Nothing you do to the pilot touches classic. +- Re-import is cheap: fix a loader mapping, drop + rebuild the pilot DB, + **delete `idmap.json`**, re-run. Nothing you do to the pilot touches classic. + The crosswalk is bound to the database it was built against and the loader + refuses a stale one, but deleting it is the habit to have - a refusal costs + you a restart, and before the guard existed a stale file silently attached + rows to the wrong assets. - Point the **collector** (GE-Enforce fleet ingest) at the pilot in parallel to confirm live PC check-ins land (see `COLLECTOR-INTEGRATION.md`), using a scoped service token. @@ -163,7 +194,7 @@ When the window is clean: 1. Freeze classic writes (announce a short read-only window). 2. Take final fresh dumps; re-run the loader into a clean pilot DB so the - cutover data is current. + cutover data is current. A clean DB means a fresh `idmap.json` too. 3. Verify counts + a fast UI spot-check. 4. Repoint the production hostname/DNS (or the reverse proxy) at the pilot. 5. Retire the classic app to read-only standby (do not delete - keep it as the diff --git a/scripts/site_imports/wjf/README.md b/scripts/site_imports/wjf/README.md index 0a2a9f09..d3351a19 100644 --- a/scripts/site_imports/wjf/README.md +++ b/scripts/site_imports/wjf/README.md @@ -33,16 +33,34 @@ resolved decisions, and remaining stages. ## Status -Complete - all 15 stages built and verified end-to-end against a fresh scratch -target (zero endpoint errors): `reference`, `employees`, `catalog`, `assets` -(the hub + machineid->assetid crosswalk), `locations`, `printers`, +All 17 stages built: `reference`, `employees`, `catalog`, `assets` (the hub + +machineid->assetid crosswalk), `locations`, `printers`, `metrology`, `communications`, `applications` (+ installs), `warranties`, `notifications`, -`knowledgebase`, `relationships`, `subnets`, `usb`, `verify`. +`knowledgebase`, `relationships`, `defaultprinters`, `subnets`, `usb`, `verify`. -Last full run: 983 assets (computer 663, machine 76, network 58, measuring-tool -136, printer 50), 24 locations, 415 employees, 850 installs, 461 primary IPs, -464 warranties, 261 notifications, 341 KB, 93 relationships, 37 subnets, 18 USB -devices + 232 events. UI spot-check passed (computer list + shop-floor map). +**The counts below are from the 2026-07-13 run and are STALE.** Two routing +changes landed after them - the `isactive=1` filter and the metrology routing +rework - so the asset mix has moved substantially. They are kept only as a +shape, not as a checklist: re-measure from the next full run before quoting +them anywhere, and do not verify a fresh import against them. + +Last full run (2026-07-13, superseded): 983 assets (computer 663, machine 76, +network 58, measuring-tool 136, printer 50), 24 locations, 415 employees, 850 +installs, 461 primary IPs, 464 warranties, 261 notifications, 341 KB, 93 +relationships, 37 subnets, 18 USB devices + 232 events. + +Known gaps, none of them blocking a scratch run: + +- `warranties`, `notifications` and `knowledgebase` POST without a duplicate + check on the API side, so resuming ACROSS one of those stages doubles its + rows. Resuming at or after them is safe. Re-run into a clean target when in + doubt. +- There is no `--dry-run` and no guard on the target: the harness mints an + admin PAT and sets `setup_complete` in whatever `DATABASE_URL` points at. + Check the target before running. +- The `verify` stage counts target `assets` against source `machines`, which is + apples to oranges - the target also holds printers and synthesized metrology + tools. Known follow-ups: general asset `locationid` is null (no source column outside printers); the ~146 duplicate machinenumbers are first-wins-skipped by decision. diff --git a/scripts/site_imports/wjf/harness.py b/scripts/site_imports/wjf/harness.py index 89438a3d..4c41542a 100644 --- a/scripts/site_imports/wjf/harness.py +++ b/scripts/site_imports/wjf/harness.py @@ -14,6 +14,7 @@ crashed run resumes and later stages resolve foreign keys. """ import json +import uuid import os import pymysql @@ -30,26 +31,63 @@ def _source_password(): with open(os.path.abspath(env)) as handle: for line in handle: if line.startswith('DATABASE_URL='): + # Not every server runs as root, and 'root:' then simply is not + # there: split()[1] raised IndexError, which read as a crash + # rather than the clear message two lines down. + if 'root:' not in line: + continue 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.""" + """Read-only access to the three source databases. - def __init__(self, host='127.0.0.1', port=3306, user='root', password=None): + The stage functions name a source by its LOGICAL name - 'shopdb_src', + 'cmmc_usb_src', 'wjf_employees_src' - and this class resolves each to a + real database. That indirection is the point: the names were hardcoded at + 29 call sites, so pointing the loader at a differently-named copy (the + prodscratch restore, or a site's own dump) meant editing run.py, which is + not something anybody should do to aim a loader. The logical names stay in + the stages where they read as documentation; the mapping lives here. + + Everything is environment-overridable, defaulting to the scratch layout: + + SOURCE_DB_MAIN default shopdb_src + SOURCE_DB_USB default cmmc_usb_src + SOURCE_DB_EMPLOYEES default wjf_employees_src + SOURCE_DB_HOST/PORT/USER/PASSWORD + """ + + # logical name -> (env var, default real name) + ALIASES = { + 'shopdb_src': ('SOURCE_DB_MAIN', 'shopdb_src'), + 'cmmc_usb_src': ('SOURCE_DB_USB', 'cmmc_usb_src'), + 'wjf_employees_src': ('SOURCE_DB_EMPLOYEES', 'wjf_employees_src'), + } + + def __init__(self, host=None, port=None, user=None, password=None): self.password = password or _source_password() - self.host, self.port, self.user = host, port, user + self.host = host or os.environ.get('SOURCE_DB_HOST', '127.0.0.1') + self.port = int(port or os.environ.get('SOURCE_DB_PORT', 3306)) + self.user = user or os.environ.get('SOURCE_DB_USER', 'root') self._conns = {} + @classmethod + def resolve(cls, name): + """The real database behind a logical source name.""" + env, default = cls.ALIASES.get(name, (None, name)) + return os.environ.get(env, default) if env else name + def db(self, name): - if name not in self._conns: - self._conns[name] = pymysql.connect( + real = self.resolve(name) + if real not in self._conns: + self._conns[real] = pymysql.connect( host=self.host, port=self.port, user=self.user, - password=self.password, database=name, charset='utf8mb4', + password=self.password, database=real, charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor) - return self._conns[name] + return self._conns[real] def rows(self, dbname, sql, args=None): with self.db(dbname).cursor() as cur: @@ -57,15 +95,62 @@ class Source: return cur.fetchall() -class IdMap: - """Persisted legacy-id -> new-id crosswalks, one namespace per entity.""" +class StaleIdMapError(RuntimeError): + """The crosswalk on disk was built against a different target database.""" - def __init__(self, path): + +class IdMap: + """Persisted legacy-id -> new-id crosswalks, one namespace per entity. + + BOUND TO ONE TARGET DATABASE. The crosswalk holds this site's legacy ids + mapped to assetids in a specific database, and those assetids mean nothing + anywhere else. Two ways that used to go wrong, both silent: + + - Delete idmap.json and re-run: every create 409s, nothing resolves into + the crosswalk, and warranties, installs, relationships and printer + assignments all quietly import as ~0 rows. + - Keep idmap.json across a drop-and-rebuild of the target: the legacy ids + now point at assetids belonging to the OLD database, so warranties and + printer assignments attach to whatever asset happens to hold that id + now. Wrong data, no error, and the runbook's two "drop and rebuild" + steps never said to delete the file. + + The bind is a marker row in the target's own settings table, so it dies + with the data: a rebuild produces a new marker, and the mismatch is caught + before a single row is written. The database NAME is not enough on its own, + because a rebuild keeps it. + """ + + MARKER_KEY = 'siteimport_targetinstance' + + def __init__(self, path, target=None): self.path = path + self.target = target self._data = {} if os.path.exists(path): with open(path) as handle: self._data = json.load(handle) + self._check_target() + + def _check_target(self): + if not self.target: + return + seen = self._data.get('_target') + if seen is None: + # Written before this check existed. Adopt it rather than refuse: + # refusing would strand a crosswalk that is probably fine, and the + # next run is bound properly. + self._data['_target'] = self.target + self.save() + return + if seen != self.target: + raise StaleIdMapError( + '%s was built against target %s, but this run targets %s.\n' + 'Those assetids belong to a different database and using them ' + 'would attach imported rows to the wrong assets.\n' + 'Delete the file to start a fresh crosswalk, or point ' + '--idmap at the one for this target.' + % (self.path, seen, self.target)) def put(self, namespace, legacy, new): self._data.setdefault(namespace, {})[str(legacy)] = new @@ -78,7 +163,8 @@ class IdMap: return str(legacy) in self._data.get(namespace, {}) def count(self, namespace): - return len(self._data.get(namespace, {})) + entry = self._data.get(namespace, {}) + return len(entry) if isinstance(entry, dict) else 0 def save(self): with open(self.path, 'w') as handle: @@ -97,11 +183,38 @@ class Harness: 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.ids = IdMap(idmap_path or default, target=self._target_instance()) self.source = Source() self.errors = [] self._defaultlevelid = None + def _target_instance(self): + """A stable id for THIS target database, created once and stored in it. + + Kept in the target's own settings table so it dies when the data does. + A drop-and-rebuild yields a new id, which is what lets IdMap catch a + crosswalk carried over from the previous database - the failure that + silently attaches imported rows to whatever asset now holds that id. + + The database name alone would not do: a rebuild keeps the name. + """ + from shopdb.core.models import Setting + existing = Setting.query.filter_by(key=IdMap.MARKER_KEY).first() + if existing and existing.value: + return existing.value + marker = uuid.uuid4().hex + if existing: + existing.value = marker + else: + db.session.add(Setting( + key=IdMap.MARKER_KEY, value=marker, valuetype='string', + category='import', + description='Identifies this database to the site-import ' + 'crosswalk. Created by the loader; do not copy it ' + 'between databases.')) + db.session.commit() + return marker + @property def defaultlevelid(self): """The level imported map positions belong to (ADR-017). diff --git a/scripts/site_imports/wjf/run.py b/scripts/site_imports/wjf/run.py index 034eabd2..43044940 100644 --- a/scripts/site_imports/wjf/run.py +++ b/scripts/site_imports/wjf/run.py @@ -9,16 +9,19 @@ Stages are idempotent (re-running resolves existing rows by natural key), so a crashed run resumes. See scripts/site_imports/wjf/harness.py for the contract this drives, and scratchpad/IMPORT-PLAN.md for the full mapping + decisions. -Implemented: reference (vendors, businessunits, operatingsystems), employees. -TODO stages (assets hub + dependents + network + usb) are stubbed - they need -the machineid->assetid crosswalk this harness persists. +All 17 stages are implemented; STAGES at the bottom of this file is the list +and main()'s --stages default is the order they must run in. The assets hub +comes before everything that references an asset, because it is what builds the +machineid->assetid crosswalk the harness persists. """ import argparse +import sys +from urllib.parse import quote import pymysql -from .harness import Harness +from .harness import Harness, Source, StaleIdMapError def _id_of(row, idfield): @@ -33,6 +36,76 @@ def _id_of(row, idfield): return None +# The server clamps any page size to MAX_PAGE_SIZE (shopdb/config.py), which +# is 100. Asking for more is not an error and not a warning - it just silently +# returns 100 rows, which is why the 409-resolve below pages instead. +PAGE_SIZE = 100 + +# A resolve that walks more pages than this is a bug in the caller, not a big +# table. Bounded so a paging mistake cannot spin forever against a live server. +MAX_RESOLVE_PAGES = 200 + + +def _rows_of(payload): + """The list out of a list-endpoint response, whichever shape it took.""" + if isinstance(payload, list): + return payload + if isinstance(payload, dict): + items = payload.get('items') + if isinstance(items, list): + return items + return [] + + +def _resolve_existing(h, lookup, unique_field, wanted, idfield): + """Find an existing row by its natural key. Returns its id, or None. + + This used to ask for ?per_page=10000 and read the single response. The + server clamps per_page to 100, so it only ever searched the first 100 rows + and returned None for everything after that. The failure was silent and + downstream: _upsert returned None, the id never reached the crosswalk, and + the run finished "successfully" having quietly dropped a model's link on + every machine that used it. The target already holds more than 100 models + and more than 100 applications, so this fired on any re-run or resume - the + exact case the loader documents as safe. + + Two passes. Some endpoints expose an exact natural-key filter for this + (?modelnumber= on models, ?appname= on applications, ?vendor= on vendors); + an endpoint without one ignores the unknown query parameter and returns its + normal first page, so the paged walk that follows is what finds the row. + """ + wanted_key = str(wanted).strip().lower() + separator = '&' if '?' in lookup else '?' + + def match(rows): + for row in rows: + if str(row.get(unique_field, '')).strip().lower() == wanted_key: + return _id_of(row, idfield) + return None + + # Pass 1: the exact filter, where the endpoint has one. + exact = '%s%s%s=%s&per_page=%d' % ( + lookup, separator, unique_field, quote(str(wanted)), PAGE_SIZE) + _, data = h.get(exact) + found = match(_rows_of(data)) + if found is not None: + return found + + # Pass 2: every page, in order. + for page in range(1, MAX_RESOLVE_PAGES + 1): + _, data = h.get('%s%sper_page=%d&page=%d' + % (lookup, separator, PAGE_SIZE, page)) + rows = _rows_of(data) + if not rows: + return None + found = match(rows) + if found is not None: + return found + if len(rows) < PAGE_SIZE: + return None + return None + + def _upsert(h, path, payload, unique_field, idfield, list_path=None): """POST payload; on 409 resolve the existing row by its unique field. Returns the new/existing id, or None.""" @@ -40,14 +113,8 @@ def _upsert(h, path, payload, unique_field, idfield, list_path=None): if status in (200, 201): return _id_of(data, idfield) if status == 409: - lookup = list_path or path - lookup += ('&' if '?' in lookup else '?') + 'per_page=10000' - _, rows = h.get(lookup) - items = rows if isinstance(rows, list) else rows.get('items', rows) - for row in (items or []): - if str(row.get(unique_field, '')).strip().lower() == \ - str(payload[unique_field]).strip().lower(): - return _id_of(row, idfield) + return _resolve_existing(h, list_path or path, unique_field, + payload[unique_field], idfield) return None @@ -418,10 +485,22 @@ def stage_metrology(h): 'businessunitid': h.ids.get('businessunit', m['businessunitid']), 'mapx': m['mapleft'], 'mapy': m['maptop'], 'levelid': h.defaultlevelid}) - if status not in (200, 201): + if status in (200, 201): + tool_assetid = _id_of(data, 'assetid') + made += 1 + elif status == 409: + # Already imported by an earlier run. Resolve it rather than + # skipping: `continue` here left the crosswalk without this tool, + # so the Controls edge below was never made and a resumed run + # produced metrology PCs driving nothing. stage_printers already + # had this branch; this stage was written later and missed it. + tool_assetid = _resolve_existing( + h, '/api/measuringtools', 'assetnumber', + f"MT-{m['machineid']}", 'assetid') + if not tool_assetid: + continue + else: continue - tool_assetid = _id_of(data, 'assetid') - made += 1 if tool_assetid and controls_id: h.ids.put('metrologytool', m['machineid'], tool_assetid) st, _ = h.post('/api/assets/relationships', { @@ -906,10 +985,27 @@ def main(): 'metrology,communications,applications,warranties,notifications,' 'knowledgebase,relationships,defaultprinters,subnets,usb,verify', help='comma list of stages to run') + parser.add_argument( + '--idmap', default=None, + help='path to the legacy-id crosswalk (default: idmap.json beside ' + 'this script). One per target database - a crosswalk from ' + 'another database maps to the wrong assets.') args = parser.parse_args() - h = Harness() - print(f'== WJ import loader (target: {h.app.config["SQLALCHEMY_DATABASE_URI"].split("/")[-1]}) ==') + try: + h = Harness(idmap_path=args.idmap) + except StaleIdMapError as error: + print('REFUSED: %s' % error) + return 1 + + target = h.app.config['SQLALCHEMY_DATABASE_URI'].split('/')[-1] + print(f'== WJ import loader (target: {target}) ==') + # Say which sources this run reads. They are overridable now, so "it + # imported nothing" is usually "it read the wrong database". + for logical in ('shopdb_src', 'cmmc_usb_src', 'wjf_employees_src'): + real = Source.resolve(logical) + if real != logical: + print(f' source {logical} -> {real}') try: for name in args.stages.split(','): name = name.strip() @@ -922,9 +1018,15 @@ def main(): print(f'\n {len(h.errors)} endpoint error(s):') for path, code, _payload, data in h.errors[:10]: print(f' {code} {path}: {data}') + # An endpoint error is not a crash, but it did leave rows unimported, + # and a run that exits 0 after 200 of them looks like a success in a + # deploy log. + if h.errors: + return 1 finally: h.close() + return 0 if __name__ == '__main__': - main() + sys.exit(main()) diff --git a/tests/test_site_import_loader.py b/tests/test_site_import_loader.py new file mode 100644 index 00000000..7b0d7f34 --- /dev/null +++ b/tests/test_site_import_loader.py @@ -0,0 +1,164 @@ +"""The WJ reference loader's resolve, crosswalk binding and source aliasing. + +The loader had no tests at all, and three of its defects were the silent kind: +they did not raise, they imported fewer rows and finished cheerfully. These +cover the logic that can be exercised without a MySQL source, which is exactly +the part that was wrong. + +Not covered here: the stage functions themselves. They need the classic source +schema, so they belong in a fixture-backed integration test against a scratch +database, not in the unit suite. +""" +import json +import os + +import pytest + +from scripts.site_imports.wjf import run as loader +from scripts.site_imports.wjf.harness import IdMap, Source, StaleIdMapError + + +class FakeHarness: + """Records the paths requested and replays canned list responses.""" + + def __init__(self, pages): + self.pages = pages + self.requested = [] + + def get(self, path): + self.requested.append(path) + return 200, self.pages.get(path, {'items': []}) + + +class TestResolveExisting: + def test_finds_a_row_beyond_the_first_page(self): + """The bug: per_page is clamped to 100, so row 101 was invisible. + + The old code asked for per_page=10000, read one response, and returned + None for anything the server did not fit in its first 100 rows. The + target already holds more than 100 models and more than 100 + applications, so this fired on every resume. + """ + page1 = {'items': [{'modelnumber': 'M%03d' % i, 'modelnumberid': i} + for i in range(100)]} + page2 = {'items': [{'modelnumber': 'LATE', 'modelnumberid': 999}]} + harness = FakeHarness({ + '/api/models?modelnumber=LATE&per_page=100': {'items': []}, + '/api/models?per_page=100&page=1': page1, + '/api/models?per_page=100&page=2': page2, + }) + found = loader._resolve_existing( + harness, '/api/models', 'modelnumber', 'LATE', 'modelnumberid') + assert found == 999 + + def test_tries_the_exact_filter_first(self): + harness = FakeHarness({ + '/api/models?modelnumber=WIDGET&per_page=100': + {'items': [{'modelnumber': 'WIDGET', 'modelnumberid': 7}]}, + }) + assert loader._resolve_existing( + harness, '/api/models', 'modelnumber', 'WIDGET', 'modelnumberid') == 7 + # One request: it must not walk every page when the filter answered. + assert len(harness.requested) == 1 + + def test_values_needing_encoding_do_not_break_the_filter(self): + harness = FakeHarness({}) + loader._resolve_existing(harness, '/api/vendors', 'vendor', + 'Acme & Co/Ltd', 'vendorid') + assert 'Acme%20%26%20Co/Ltd' in harness.requested[0] or \ + 'Acme%20%26%20Co%2FLtd' in harness.requested[0] + + def test_matching_ignores_case_and_padding(self): + harness = FakeHarness({ + '/api/vendors?vendor=acme&per_page=100': + {'items': [{'vendor': ' ACME ', 'vendorid': 3}]}, + }) + assert loader._resolve_existing( + harness, '/api/vendors', 'vendor', 'acme', 'vendorid') == 3 + + def test_missing_row_returns_none_without_spinning(self): + harness = FakeHarness({}) + assert loader._resolve_existing( + harness, '/api/models', 'modelnumber', 'NOPE', 'modelnumberid') is None + # Empty first page ends it: the filter attempt plus one page. + assert len(harness.requested) == 2 + + def test_a_short_page_ends_the_walk(self): + harness = FakeHarness({ + '/api/models?per_page=100&page=1': {'items': [{'modelnumber': 'A', 'modelnumberid': 1}]}, + }) + assert loader._resolve_existing( + harness, '/api/models', 'modelnumber', 'ZZZ', 'modelnumberid') is None + assert len(harness.requested) == 2 + + def test_a_bare_list_response_is_accepted(self): + harness = FakeHarness({ + '/api/x?name=a&per_page=100': [{'name': 'a', 'id': 5}], + }) + assert loader._resolve_existing(harness, '/api/x', 'name', 'a', 'id') == 5 + + def test_an_existing_query_string_keeps_its_separator(self): + harness = FakeHarness({}) + loader._resolve_existing(harness, '/api/models?category=pc', + 'modelnumber', 'M1', 'modelnumberid') + assert harness.requested[0].startswith('/api/models?category=pc&modelnumber=') + + +class TestIdMapBinding: + def test_a_crosswalk_from_another_database_is_refused(self, tmp_path): + """The silent corruption this guard exists for. + + Keeping idmap.json across a drop-and-rebuild leaves legacy ids pointing + at assetids from the OLD database, so warranties and printer + assignments attach to whatever asset holds that id now. No error, wrong + data. + """ + path = tmp_path / 'idmap.json' + path.write_text(json.dumps({'_target': 'aaaa', 'asset': {'1': 10}})) + with pytest.raises(StaleIdMapError) as caught: + IdMap(str(path), target='bbbb') + assert 'different database' in str(caught.value) + + def test_the_matching_target_loads(self, tmp_path): + path = tmp_path / 'idmap.json' + path.write_text(json.dumps({'_target': 'aaaa', 'asset': {'1': 10}})) + ids = IdMap(str(path), target='aaaa') + assert ids.get('asset', 1) == 10 + + def test_an_unstamped_crosswalk_is_adopted_not_refused(self, tmp_path): + # Written before the guard existed. Refusing would strand a crosswalk + # that is probably fine; adopting binds it for every run after. + path = tmp_path / 'idmap.json' + path.write_text(json.dumps({'asset': {'1': 10}})) + ids = IdMap(str(path), target='ccc') + assert ids.get('asset', 1) == 10 + assert json.loads(path.read_text())['_target'] == 'ccc' + + def test_the_bookkeeping_key_is_not_counted_as_a_namespace(self, tmp_path): + path = tmp_path / 'idmap.json' + ids = IdMap(str(path), target='ccc') + ids.put('asset', 1, 10) + assert ids.count('asset') == 1 + assert ids.count('_target') == 0 + + def test_a_fresh_crosswalk_needs_no_stamp_to_start(self, tmp_path): + ids = IdMap(str(tmp_path / 'new.json'), target='ddd') + ids.put('asset', 5, 50) + assert ids.get('asset', 5) == 50 + + +class TestSourceAliasing: + def test_logical_names_default_to_the_scratch_databases(self, monkeypatch): + for var in ('SOURCE_DB_MAIN', 'SOURCE_DB_USB', 'SOURCE_DB_EMPLOYEES'): + monkeypatch.delenv(var, raising=False) + assert Source.resolve('shopdb_src') == 'shopdb_src' + assert Source.resolve('cmmc_usb_src') == 'cmmc_usb_src' + assert Source.resolve('wjf_employees_src') == 'wjf_employees_src' + + def test_the_environment_repoints_the_loader(self, monkeypatch): + """What unblocks the prodscratch run without editing run.py.""" + monkeypatch.setenv('SOURCE_DB_MAIN', 'prodscratch') + assert Source.resolve('shopdb_src') == 'prodscratch' + + def test_an_unknown_name_passes_through(self, monkeypatch): + assert Source.resolve('something_else') == 'something_else'