mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
fix(backup): preserve NOT NULL/DEFAULT/FK/UNIQUE in Postgres→SQLite backup (#2526)
On a PostgreSQL install, create_backup_zip() exports a portable SQLite copy so backups move between engines. It rebuilt each table with only column name + type + PK, dropping NOT NULL, server_default/DEFAULT, foreign keys, and unique constraints. Restore onto SQLite page-copies that schema straight onto the live database, and post-restore init_db() can't repair it (create_all is CREATE TABLE IF NOT EXISTS). So server_default columns like spoolbuddy_devices.created_at (server_default=func.now()) ended up with no DEFAULT: SQLAlchemy omits them on INSERT, the DB wrote NULL, and the next read 500'd on Pydantic validation. Every server_default column was exposed the same way; the FK/unique loss followed from the same simplified CREATE TABLE. Build the portable schema with Base.metadata.create_all() against a SQLite engine instead of the hand-rolled loop, so it emits the exact DDL a native SQLite install gets (NOT NULL, DEFAULT func.now() -> CURRENT_TIMESTAMP, FKs, unique constraints, indexes). The data-export insert path is unchanged, and the #1333 OIDC-icon guard is preserved automatically (LargeBinary -> BLOB), which lets the now-redundant _sqlalchemy_type_to_sqlite_type() helper be removed. Fixes newly-created backups; a backup from an older build still carries the degraded schema, so re-take backups after upgrading. Replace the #1333 type-mapping unit tests with three that inspect the real backup schema via metadata.create_all + PRAGMA table_info: icon_data is BLOB, created_at keeps its CURRENT_TIMESTAMP DEFAULT, a NOT NULL non-PK column stays NOT NULL.
This commit is contained in:
maziggy
@@ -37,6 +37,7 @@ All notable changes to Bambuddy will be documented in this file.
|
||||
- **Sort File Manager folder tree by recent activity (#1770, requested by @Kingbuzz0)** — Until now the folder tree was always sorted alphabetically by name, both backend (`order_by(LibraryFolder.name)`) and frontend. The reporter — a user with a lot of nested cad / slicer directories — wanted "find folders that just got a new 3MF" without scrolling the whole alphabet. **What changed.** The folder sidebar header gains a small dropdown (**By name** / **By recent activity**) plus an asc / desc arrow button, sitting alongside the existing Collapse + Wrap toggles. Choice persists per-browser via `localStorage` (`library-folder-sort-field`, `library-folder-sort-direction`) so the preference survives reloads. **Activity semantics.** `latest_activity_at` per folder = `MAX(folder.updated_at, MAX(immediate-child file.updated_at))`. The DB had the data — `LibraryFile.updated_at` is `onupdate=func.now()` and `LibraryFolder.updated_at` the same — but `LibraryFolder.updated_at` alone only bumps on rename / move, not on file-add inside the folder, which is exactly the wrong signal for "did I just drop a new model in here." The aggregate fixes that. Recursion across subfolders is intentionally **NOT** computed — a deeply nested new 3MF bubbles its immediate parent, not every ancestor up to the root. This keeps the route a single `GROUP BY` rather than a recursive CTE, matching the existing file_counts subquery shape sibling at `library.py:746`. A future Tier 3 follow-up could add the recursive-CTE variant if anyone reports deeply-nested updates not bubbling far enough. **Backend.** New `latest_activity_at: datetime | None` field on `FolderResponse` and `FolderTreeItem` schemas. The `/folders` tree route picks up a sibling `func.max(LibraryFile.updated_at)` group-by alongside the existing file-count subquery; resolves the field per row. The `/folders/by-project/{id}` and `/folders/by-archive/{id}` routes collapse their per-row file-count subquery to fetch `count + max` in one trip (one extra column, zero extra round-trips). All 5 single-folder constructors (POST `/folders`, GET `/folders/{id}`, PUT `/folders/{id}`, POST `/folders/external`, the create flows) populate the field with `max(folder.updated_at, latest_file)` or fall back to `folder.updated_at` when there are no files, so the API surface is consistent across every route that returns a folder. **External folders.** `LibraryFile` rows are created for scanned external files too (`library.py:526`), so the MAX aggregate works on them — but the timestamp reflects when Bambuddy last *scanned / re-indexed* the file, not the filesystem mtime. For a NAS that gets new files added outside Bambuddy, the activity-sort lags until the next scan. Documented in the file-manager wiki page rather than papered over with `os.stat()` on every list call, which would stall the route on slow mounts. **Frontend.** A new recursive `sortedFolders` `useMemo` applies the comparator uniformly to top-level + every nested `children` level so sort order is consistent at every depth. Comparator falls back to name when activity timestamps tie or are both null, so an empty folder never elbows a recently-used one to a random place — empties go to the end of the activity bucket regardless of direction. Both the desktop sidebar render and the mobile selector dropdown consume `sortedFolders` so the order is identical across breakpoints. The single-folder `findFolder()` traversal and `selectedFolder` memo still operate on the unsorted `folders` because they index by ID — sort-order-independent. **Recursion safety.** The sort creates fresh object refs at every level on every memo invocation; the `FolderTreeItem` keys stay ID-based (`${folder.id}-${collapseFoldersByDefault ? 'c' : 'e'}`) so React reconciliation by ID preserves folder expansion state across sort flips. **i18n.** 3 new keys in `fileManager.*` (`folderSort`, `folderSortByName`, `folderSortByActivity`) translated in all 11 locales (de / en / es / fr / it / ja / ko / pt-BR / tr / zh-CN / zh-TW), no English fallback. Parity 5238 leaves per locale. **Tests.** 2 new backend integration cases in `test_library_api.py` (file-in-folder bubbles `latest_activity_at` to the file's timestamp, empty folder falls back to `folder.updated_at`). All 152 library + folder + trash + slice integration tests still pass; 51/51 FileManagerPage frontend tests still pass; 26/26 QueuePage tests still pass; `npm run build` clean; `ruff` clean; i18n parity green.
|
||||
|
||||
### Fixed
|
||||
- **Postgres→SQLite backup dropped NOT NULL / DEFAULT / FK / UNIQUE, causing NULLs after restore (#2526, reporter @bmorrison9)** — On a PostgreSQL install, `create_backup_zip()` exports a portable SQLite copy of the database so backups can move between engines. It rebuilt each table with only column name + type + primary key — the code's own comment admitted "simplified — just column names and types" — dropping `NOT NULL`, `server_default`/`DEFAULT`, foreign keys, and unique constraints. When such a backup is restored onto a SQLite install, `restore_backup()` page-copies the file straight onto the live database (`sqlite3.Connection.backup()`), so the stripped-down schema becomes the *running* database; the post-restore `init_db()` can't repair it because `create_all()` is `CREATE TABLE IF NOT EXISTS` and never alters existing tables. The reporter root-caused it precisely: `SpoolBuddyDevice.created_at` is `server_default=func.now()`, so SQLAlchemy omits the column on INSERT and relies on the DB default — but with no `DEFAULT` clause the row got a bare `NULL`, which then failed Pydantic validation (`DeviceResponse.created_at`) on the next read and 500'd. Every `server_default` column across the schema was exposed the same way, and the FK/unique loss followed from the same simplified CREATE TABLE. **Fix.** The PostgreSQL branch now builds the portable SQLite schema with `Base.metadata.create_all()` against a SQLite engine — the exact DDL a native SQLite install gets — instead of the hand-rolled loop. That emits `NOT NULL`, `DEFAULT` (`server_default=func.now()` → `DEFAULT (CURRENT_TIMESTAMP)`), foreign keys, unique constraints, and indexes, so a Postgres→SQLite restore reproduces the same effective schema a fresh SQLite install would have. The data-export insert path is unchanged, and the `#1333` OIDC-icon guard is preserved automatically — `LargeBinary` renders as `BLOB` under the real DDL — which let the now-redundant `_sqlalchemy_type_to_sqlite_type()` type-mapping helper be removed. This fixes newly-created backups; a backup taken with an older build still carries the degraded schema, so re-take backups after upgrading. **Tests.** The `#1333` type-mapping unit tests were replaced with three that inspect the *real* backup schema (`metadata.create_all` on SQLite, read back via `sqlite_master`/`PRAGMA table_info`): the OIDC `icon_data` column is `BLOB` (#1333), `spoolbuddy_devices.created_at` keeps its `CURRENT_TIMESTAMP` DEFAULT (#2526), and a NOT NULL non-PK column stays NOT NULL. Full backup/restore suite (`test_settings_api`, `test_security`, Postgres-restore-cascade, SQLite-WAL-safety, OIDC-blob-roundtrip) 136/136 green; `ruff` clean. **Scope.** Backend-only, PostgreSQL-source backups. No DB migration, no new permission, no i18n change.
|
||||
- **"Store on external storage" diagnostic reported an unresolvable fail on P1S/P1P (#2524, reporter @gregspatrick)** — Install-step-4's `external_storage` check hard-**failed** for a P1S even though there is no reachable UI anywhere — Bambu Studio, OrcaSlicer, Handy, or the printer (P1S has no screen) — that can turn the option on. The reporter root-caused it precisely: `has_external_storage()` returns True for the P1S (it does have a MicroSD slot), so the check proceeds to read `state.store_to_sdcard` (MQTT `home_flag` bit 11), which is stuck `False`. The toggle only renders in Studio when the printer publishes `support_save_remote_print_file_to_storage`, and current P1-series firmware (through 01.10.00.00) never does — Bambu's own storage-cache wiki lists P1 Series as "Not Supported". So the user was shown a red fail they could never clear. **Fix.** New `NO_REMOTE_STORAGE_TOGGLE_MODELS` set (P1S, P1P) + `has_remote_storage_toggle()` helper, distinct from the no-slot `NO_EXTERNAL_STORAGE_MODELS` used for A1/A1 Mini (the P1S genuinely *has* a slot — conflating the two would be wrong). When a model has a slot but no reachable toggle and `store_to_sdcard` is False, the diagnostic now emits `skip` with `params={"reason": "unsupported_model"}` instead of `fail`, and the overall result no longer escalates to "problems" for it. A P1S that somehow reports the option *on* still passes. The gate is model-scoped and default-open, so X1/P2S/H2-class printers — where the toggle is reachable and the fail is actionable — are unaffected; if a future P1 firmware surfaces the capability, drop the model from the set and the check reactivates. The frontend `DiagnosticChecklist` picks a reason-specific message variant (`external_storage.skip_unsupported_model`) when a check carries a `reason`, falling back to the plain per-status text otherwise — so instead of the misleading generic "needs a live MQTT connection" skip line, P1 users see an accurate explanation that the option can't be enabled on current firmware and archived prints may lack thumbnails/metadata until Bambu adds support. **i18n.** 1 new key translated across all 11 locales; parity green at 5580 leaves each. **Tests.** Backend: 3 diagnostic cases (P1S/P1P → skip with the reason param, overall stays "ok"; P1S with the option on still passes) + 3 helper cases in `test_printer_models.py` (P1-series false, other models/unknown/empty true). Frontend: 2 `ConnectionDiagnosticModal` cases (reason variant renders and suppresses the generic text; no-reason falls back). Full diagnostic + model suites green; `ruff` clean; `npm run build` + ESLint clean. **Scope.** No DB migration, no new permission.
|
||||
- **Finish photo still caught the swapped/empty plate intermittently on A1 Mini + SwapMod (#1867 follow-on, reporter @qoatzelcoat)** — The last-layer edge trigger shipped in 0.2.4.9 fixed most cases but the reporter still saw the wrong (post-swap) plate now and then — "nothing changed, just kept adding files to the queue." Root cause, confirmed from the support bundle: this A1 Mini firmware (01.08.01.00) **never emits `stg_cur=22`** — across the whole 35k-line log (24 completions) the only stages it reports are 0/2/3/4/13/14/54/77/255, so every completion falls through to the `FINISH`-state fallback. Bambu only reports `gcode_state=FINISH` *after* the user End G-code runs (the print's last object layer was laid ~2 min before FINISH), so a live grab there is guaranteed to show the SwapMod-ejected plate. The last-layer edge (`layer_num >= total_layer_num` while RUNNING) is the right window but depends on catching **one transient MQTT packet** — if the firmware coalesces or drops the final `layer_num == total` push and jumps straight to FINISH, the edge is missed and it silently reverts to the post-swap grab. That's the intermittency; queued prints run unattended so the misses accumulate. **Fix — bank a frame instead of chasing an edge.** Bambuddy now keeps a rolling "last in-print camera frame" per printer, refreshed on layer change (throttled to ~25 s, always refreshed on the final object layer) via the same snapshot path the finish photo uses — so it honours the `capture_finish_photo` setting and works for external cameras, buffered RTSP, and fresh RTSP grabs alike. Because banking is **layer-driven it freezes automatically the instant printing ends**: the End G-code (plate swap) emits no further `layer_num` increases, so the last banked frame is always the finished print before the swap. On the `FINISH`-state fallback the finish-photo path now prefers the banked frame over a live grab; the `stage_22` and `last_layer` triggers still live-grab (they fire before the swap and give cleaner parked-toolhead framing), and if no banked frame exists it degrades to the old live grab rather than sending a text-only notification. Correctness no longer depends on *which* signal the firmware emits or on catching the edge — a missed edge just means the photo is one layer-frame stale (a finished print, not an empty plate). The bank is cleared on print start so a queued job can't reuse the prior job's frame. **Tests.** 8 new cases in `test_finish_photo_moment_sync.py`: `finish_state` prefers the banked frame and skips the live grab, falls back to live when no bank exists, and `last_layer` ignores the bank; plus 5 for the banking helper — stores while printing, throttles within the interval, always refreshes on the last layer, skips when not RUNNING (the freeze), and skips during calibration sub-stages. Full finish-photo + MQTT + layer-timelapse suites 355/355 green; `ruff check` clean. **Scope.** Backend-only. No DB migration, no new permission, no i18n change.
|
||||
- **P1/A1 camera stayed black on load until a ~20-minute self-heal (#2521, reporter @nnimby848)** — On chamber-image printers (P1S/A1, port 6000) the camera view frequently came up black on every load/reload and only recovered ~20 min later. The reporter supplied excellent evidence — backend logs, tcpdump (frames actively flowing on port 6000), and a HAR — and correctly identified the trigger: the viewer mounts twice in quick succession (React StrictMode + a self-inflicted reconnect loop), so a short-lived first viewer attaches and detaches within tens of ms while a second viewer persists. Their proposed mechanism (the connection being "attributed" to the dead viewer's trace ID) was a misread of the architecture — `camera_fanout.py` is a **shared** fan-out (one upstream socket per printer, keyed `{id}-fanout`), so only the first subscriber logs "Starting/connected" and every later viewer taps the same pump; the trace ID in the log is just `contextvars` context, and the HAR `status:0` is a mid-stream capture artifact, not a missing response. The real defects were two, both real: **(1) Late-subscriber cold-start.** Every viewer after the first got a fresh empty queue and had to wait for the *next* upstream frame; on a slow chamber cam plus the churn the `<img>` never fired `onLoad`, so the page's stall-detector reconnected every few seconds, minting yet another short-lived subscriber — a self-sustaining loop. **(2) Single-connection socket overlap.** Port 6000 allows one connection; the churn tore the upstream down and reopened it, and a replacement broadcaster could open a **new** socket before the old one finished closing (`_grace_then_stop` exposed `stopped=True` before the pump's socket-close `finally` completed). The printer kept feeding the orphaned socket and starved the live one until its TCP keepalive reaped it — the ~20 min self-heal. **Fixes.** *Backend fan-out:* the broadcaster now remembers the last chunk it pumped and **primes a late/surviving subscriber with it on `subscribe()`**, so any viewer after the first renders a frame instantly (fires `onLoad`, resets the reconnect loop, ends the churn); and a replacement broadcaster's **pump now waits for the displaced broadcaster's upstream socket to fully close** (`wait_until_torn_down()`, set only after the pump's cancellation + socket-close `finally`) before it dials the printer, so two sockets to a single-connection printer never overlap. Guarding at the pump rather than at `get_or_create_broadcaster` keeps it correct when concurrent viewers race to replace the same stopped broadcaster — only the single pump dials — and it's bounded by a 10 s cap so a wedged close degrades to the old behaviour instead of never producing a frame. *Frontend (`CameraPage`):* the stall-detector now requires **two consecutive** stalled/inactive status reads (~10 s) before reconnecting, so a single blip during fan-out startup/handover no longer nukes a stream that's about to deliver frames; the strike counter resets on a rendered frame and on each fresh load. **Tests.** 6 new fan-out unit cases in `test_camera_fanout.py` — late subscriber primed with last frame, first subscriber not primed, `wait_until_torn_down` completes after shutdown, the replacement barrier blocks until the prior teardown completes, and the barrier's bounded-timeout fallback. Full camera suite (fan-out + `test_camera_api` + stderr-summary) 70/70 green; `ruff check` clean; frontend `npm run build` + ESLint clean; existing 13 `CameraPage` cases stay green. **Scope.** No DB migration, no new permission, no new i18n key. The single-connection socket-overlap fix also benefits any single-camera-slot model (e.g. X2D on firmware that permits one connection). If the black screen ever persists on a specific firmware, a per-frame debug counter or `ss -tn | grep :6000` during the episode would confirm whether the upstream is delivering frames — but priming + teardown discipline address both observed mechanisms.
|
||||
|
||||
@@ -35,32 +35,6 @@ _SENSITIVE_FIELDS_FOR_API_KEY = (
|
||||
)
|
||||
|
||||
|
||||
def _sqlalchemy_type_to_sqlite_type(type_repr: str) -> str:
|
||||
"""Map a SQLAlchemy column type's ``str()`` to a SQLite-native column type.
|
||||
|
||||
Used by ``create_backup_zip`` to reconstruct a portable SQLite database
|
||||
file from PostgreSQL data. Falling through to TEXT for binary columns
|
||||
corrupts non-UTF8 bytes — the BLOB branch is the #1333 regression guard
|
||||
for OIDC icon BLOBs.
|
||||
|
||||
Extracted as a pure helper so it can be unit-tested without spinning up
|
||||
the full FastAPI app + backup pipeline.
|
||||
"""
|
||||
type_str = type_repr.upper()
|
||||
if "INT" in type_str:
|
||||
return "INTEGER"
|
||||
if "FLOAT" in type_str or "REAL" in type_str or "NUMERIC" in type_str:
|
||||
return "REAL"
|
||||
if "BOOL" in type_str:
|
||||
return "BOOLEAN"
|
||||
if "BLOB" in type_str or "BYTEA" in type_str or "BINARY" in type_str:
|
||||
# OIDC icon BLOB column (#1333) — without this branch the column
|
||||
# was created as TEXT and non-UTF8 bytes were corrupted during the
|
||||
# PG→SQLite-ZIP backup round trip.
|
||||
return "BLOB"
|
||||
return "TEXT"
|
||||
|
||||
|
||||
async def get_setting(db: AsyncSession, key: str) -> str | None:
|
||||
"""Get a single setting value by key."""
|
||||
result = await db.execute(select(Settings).where(Settings.key == key))
|
||||
@@ -582,25 +556,31 @@ async def create_backup_zip(output_path: Path | None = None) -> tuple[Path, str]
|
||||
import json
|
||||
import sqlite3
|
||||
|
||||
from sqlalchemy import create_engine as create_sync_engine
|
||||
|
||||
from backend.app.core.database import Base, engine
|
||||
|
||||
backup_db_path = temp_path / "bambuddy.db"
|
||||
dst = sqlite3.connect(str(backup_db_path))
|
||||
metadata = Base.metadata
|
||||
|
||||
# Create tables in SQLite backup (simplified — just column names and types)
|
||||
for table in metadata.sorted_tables:
|
||||
cols = []
|
||||
pk_cols = [col.name for col in table.columns if col.primary_key]
|
||||
for col in table.columns:
|
||||
col_type = _sqlalchemy_type_to_sqlite_type(str(col.type))
|
||||
# Only inline PRIMARY KEY for single-column PKs
|
||||
pk = " PRIMARY KEY" if col.primary_key and len(pk_cols) == 1 else ""
|
||||
cols.append(f"{col.name} {col_type}{pk}")
|
||||
# Add composite primary key constraint if needed
|
||||
if len(pk_cols) > 1:
|
||||
cols.append(f"PRIMARY KEY ({', '.join(pk_cols)})")
|
||||
dst.execute(f"CREATE TABLE IF NOT EXISTS {table.name} ({', '.join(cols)})") # noqa: S608
|
||||
# Build the portable SQLite schema with SQLAlchemy's own DDL rather
|
||||
# than a hand-rolled CREATE TABLE. metadata.create_all() emits the
|
||||
# exact schema a native SQLite install gets — NOT NULL, DEFAULT
|
||||
# (server_default=func.now() → CURRENT_TIMESTAMP), foreign keys,
|
||||
# unique constraints and indexes. The previous name+type-only
|
||||
# rebuild dropped all of these, so a Postgres→SQLite restore left
|
||||
# server_default columns (e.g. spoolbuddy_devices.created_at) with
|
||||
# no DEFAULT — SQLAlchemy omits such columns on INSERT and the DB
|
||||
# then wrote NULL, which 500'd on the next read (#2526). Using the
|
||||
# real DDL also keeps the #1333 BLOB guard: LargeBinary still
|
||||
# renders as BLOB, so OIDC icon bytes survive the round trip.
|
||||
schema_engine = create_sync_engine(f"sqlite:///{backup_db_path}")
|
||||
try:
|
||||
metadata.create_all(schema_engine)
|
||||
finally:
|
||||
schema_engine.dispose()
|
||||
|
||||
dst = sqlite3.connect(str(backup_db_path))
|
||||
|
||||
# Export data from Postgres to SQLite
|
||||
async with engine.connect() as conn:
|
||||
|
||||
@@ -1,54 +1,85 @@
|
||||
"""Type-mapping coverage for the OIDC icon BLOB column (#1333).
|
||||
"""Backup-schema fidelity for the PG→SQLite portable export (#1333, #2526).
|
||||
|
||||
Bambuddy's ``create_backup_zip`` rebuilds the SQLite backup schema from
|
||||
``Base.metadata`` when the source database is PostgreSQL. The column-type
|
||||
mapping previously fell through to ``TEXT`` for any unknown SQLAlchemy
|
||||
type — including ``LargeBinary`` / ``BYTEA`` — which corrupts non-UTF8
|
||||
icon bytes during the PG → SQLite-ZIP round trip.
|
||||
Bambuddy's ``create_backup_zip`` rebuilds the SQLite backup schema when the
|
||||
source database is PostgreSQL. It now uses ``Base.metadata.create_all()``
|
||||
against a SQLite engine — the same DDL a native SQLite install gets — rather
|
||||
than a hand-rolled ``name + type`` CREATE TABLE. The old rebuild dropped two
|
||||
things that these tests pin:
|
||||
|
||||
These tests exercise the extracted ``_sqlalchemy_type_to_sqlite_type``
|
||||
helper directly so the regression guard doesn't depend on a full backup
|
||||
pipeline. The SQLite source path is just ``shutil.copy2`` of the live
|
||||
.db file and is therefore unaffected by the type mapping.
|
||||
* ``LargeBinary`` fell through to ``TEXT``, corrupting non-UTF8 OIDC icon
|
||||
bytes during the round trip (#1333). ``create_all`` renders it as ``BLOB``.
|
||||
* ``NOT NULL`` / ``DEFAULT`` / FK / ``UNIQUE`` were all dropped, so a
|
||||
Postgres→SQLite restore left ``server_default`` columns (e.g.
|
||||
``spoolbuddy_devices.created_at``) with no ``DEFAULT`` — later inserts
|
||||
wrote ``NULL`` and 500'd on read (#2526). ``create_all`` emits the default.
|
||||
|
||||
The SQLite *source* path is just ``shutil.copy2`` of the live .db file and is
|
||||
therefore unaffected — these guards only matter for the PostgreSQL branch.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import sqlite3
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import Column, LargeBinary
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from backend.app.api.routes.settings import _sqlalchemy_type_to_sqlite_type
|
||||
from backend.app.core.database import Base
|
||||
from backend.tests._fixtures.oidc_icon import PNG_BYTES as _PNG_BYTES
|
||||
|
||||
|
||||
class TestTypeMapping:
|
||||
"""Unit-level coverage of the helper that backups use for PG→SQLite."""
|
||||
def _build_backup_schema(db_path) -> dict[str, dict]:
|
||||
"""Build the portable SQLite schema exactly as create_backup_zip's
|
||||
PostgreSQL branch does, then return ``{table: {col: PRAGMA row}}``.
|
||||
|
||||
def test_largebinary_maps_to_blob(self):
|
||||
# Direct from a SQLAlchemy LargeBinary column — this is exactly
|
||||
# what the create_backup_zip loop calls str() on.
|
||||
col = Column(LargeBinary)
|
||||
assert _sqlalchemy_type_to_sqlite_type(str(col.type)) == "BLOB"
|
||||
PRAGMA table_info rows are ``(cid, name, type, notnull, dflt_value, pk)``.
|
||||
"""
|
||||
engine = create_engine(f"sqlite:///{db_path}")
|
||||
try:
|
||||
Base.metadata.create_all(engine)
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"type_repr",
|
||||
["BLOB", "BYTEA", "BYTEA(1024)", "VARBINARY", "BINARY", "binary varying"],
|
||||
)
|
||||
def test_binary_type_strings_map_to_blob(self, type_repr):
|
||||
assert _sqlalchemy_type_to_sqlite_type(type_repr) == "BLOB"
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
try:
|
||||
schema: dict[str, dict] = {}
|
||||
tables = [
|
||||
row[0]
|
||||
for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")
|
||||
]
|
||||
for table in tables:
|
||||
schema[table] = {row[1]: row for row in conn.execute(f"PRAGMA table_info({table})")} # noqa: S608
|
||||
return schema
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def test_integer_unchanged(self):
|
||||
assert _sqlalchemy_type_to_sqlite_type("INTEGER") == "INTEGER"
|
||||
assert _sqlalchemy_type_to_sqlite_type("BIGINT") == "INTEGER"
|
||||
|
||||
def test_boolean_unchanged(self):
|
||||
assert _sqlalchemy_type_to_sqlite_type("BOOLEAN") == "BOOLEAN"
|
||||
class TestBackupSchemaFidelity:
|
||||
"""The real backup-schema builder (metadata.create_all on SQLite),
|
||||
inspected via sqlite_master, keeps the constraints the old name+type
|
||||
rebuild dropped."""
|
||||
|
||||
def test_unknown_falls_back_to_text(self):
|
||||
assert _sqlalchemy_type_to_sqlite_type("VARCHAR(500)") == "TEXT"
|
||||
assert _sqlalchemy_type_to_sqlite_type("DATETIME") == "TEXT"
|
||||
def test_icon_data_column_is_blob(self, tmp_path):
|
||||
# #1333 — LargeBinary must render as BLOB, not TEXT, or non-UTF8
|
||||
# OIDC icon bytes are corrupted on the PG→SQLite round trip.
|
||||
schema = _build_backup_schema(tmp_path / "schema.db")
|
||||
assert schema["oidc_providers"]["icon_data"][2] == "BLOB"
|
||||
|
||||
def test_server_default_column_keeps_default(self, tmp_path):
|
||||
# #2526 — a server_default=func.now() column must carry a DEFAULT so
|
||||
# inserts that omit it (SQLAlchemy does, for server-side defaults)
|
||||
# don't write NULL after a Postgres→SQLite restore.
|
||||
schema = _build_backup_schema(tmp_path / "schema.db")
|
||||
created_at = schema["spoolbuddy_devices"]["created_at"]
|
||||
assert created_at[4] is not None, "created_at lost its DEFAULT clause"
|
||||
assert "CURRENT_TIMESTAMP" in str(created_at[4]).upper()
|
||||
|
||||
def test_not_null_column_keeps_not_null(self, tmp_path):
|
||||
# #2526 — NOT NULL columns must stay NOT NULL. A single-column PK is
|
||||
# implicitly NOT NULL, so assert on a non-PK required column.
|
||||
schema = _build_backup_schema(tmp_path / "schema.db")
|
||||
# notnull flag is index 3 of the PRAGMA row.
|
||||
assert schema["spoolbuddy_devices"]["device_id"][3] == 1
|
||||
|
||||
|
||||
class TestSqliteBinaryRoundtrip:
|
||||
|
||||
Reference in New Issue
Block a user