public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-09-30 11:12:35 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
3bb087db54ef5eb90f3dc3a43af7b3d0ea59eaff
bambuddy/backend/tests/unit/test_github_restore.py
T
jmoore-skild 3bb087db54 fix(backup): report the rows a failed K-profile step already committed (#2656) 
`_apply` commits the database categories before the K-profile phase, and the
comment there is right about why: `get_kprofiles` is 3 x 5 s per printer per
nozzle and SQLite's `busy_timeout` is 15 s, so holding the writer across the
MQTT phase would fail every concurrent writer in the app.

But `run_restore`'s handler returns `{"success": False, ..., "results": {}}`
for anything raised after that point, and the per-call guards inside
`_restore_kprofiles` do not cover the whole phase. Two consequences, and the
second is worse:

* The user is told the restore failed and handed an empty `results` while the
  archive, spool and settings rows are durable on disk. The honest-reporting
  theme this whole feature is built on inverted on exactly the path where it
  matters most.
* `_reconfigure_mqtt_relay` sits inside the same `try`, downstream of the
  raise. A restore that rewrote the mqtt_* rows left the relay pointed at the
  pre-restore broker until something else reconfigured it.

`_apply` now contains the K-profile phase: fold the error into that category's
tally as `failed` plus a `kprofilesStepFailed` note, and let the results it has
already committed be returned and reported. Every profile the payload carried
and the phase did not account for is counted failed — silence would have been
the same lie in a smaller font. `_reconfigure_mqtt_relay` is reached again
because `_apply` returns normally. The rollback in the handler discards only
the phase's own read transaction, so a database error cannot leave the session
in a state that turns the caller's commit into the very report this prevents.

`kprofilesSendFailed` was the obvious note to reuse and is the wrong one: it
names a nozzle, a printer and a serial that a phase-level failure does not
have, and "failed to send" is untrue of a step that never got as far as
sending. One new leaf x 13 locales instead.

Belt-and-braces on the trigger that found this:
`sum(len(c.get("profiles") or []) ...)` raises TypeError on a hand-edited or
truncated backup whose `profiles` is not a list, and it runs before the guards.
Counting defensively makes that a skipped category rather than an exception
thrown over committed rows.

Control kept explicit: a failure *before* the commit still rolls back, still
reports nothing restored, and still does not touch the relay.

Tests: +5 (280 -> 285 across the three restore files, 328 -> 337 across
`-k github`). Fail-pre-fix 4 — 3 for the containment, 1 for the defensive
count, checked separately. i18n parity 13 locales at 5771 leaves.

Bundle rebuilt for the new leaf: index-CHCEEMgx.js -> index-DhOfNgMz.js. CSS
hash unchanged.
2026-08-04 08:57:34 -04:00

2573 lines
112 KiB
Python
Raw Blame History

"""Unit tests for the Git backup restore service (#2656).
Focus is on the per-category appliers: natural-key matching, the deliberate
refusal to reuse the backup's primary keys, old_id -> new_id remapping for
dependent rows, overwrite-vs-skip, the settings credential blocklist, and the
K-profile paths that depend on live printers.
"""
from datetime import datetime
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from sqlalchemy import select
from backend.app.models.archive import PrintArchive
from backend.app.models.settings import Settings
from backend.app.models.spool import Spool
from backend.app.models.spool_usage_history import SpoolUsageHistory
from backend.app.models.user import User
from backend.app.schemas.github_backup import GitHubRestoreRequest, RestoreCategory
from backend.app.services.github_restore import (
_COMPANION_CREDENTIAL_ENV,
_COMPANION_CREDENTIALS,
_COMPANION_EXPOSURE_TOGGLES,
ARCHIVES_PATH,
SETTINGS_PATH,
SPOOL_USAGE_PATH,
SPOOLS_PATH,
GitHubRestoreService,
_CategoryTally,
_is_blocked_setting_key,
_is_protected_setting_key,
_is_usable_credential,
_parse_dt,
_setting_value_is_true,
_SettingsPlan,
)
def _service() -> GitHubRestoreService:
return GitHubRestoreService()
def _messages(tally: _CategoryTally) -> list[str]:
"""The English rendering of each note.
Notes are ``{code, params, message}`` since they became translatable
(#2656); asserting on the message keeps these tests readable while
``_codes`` covers the half a client actually keys on.
"""
return [note["message"] for note in tally.notes]
def _codes(tally: _CategoryTally) -> list[str]:
return [note["code"] for note in tally.notes]
class TestParseDt:
def test_parses_str_datetime_the_backup_writes(self):
assert _parse_dt("2026-07-27 06:02:05.123456") == datetime(2026, 7, 27, 6, 2, 5, 123456)
def test_parses_iso_with_t_separator(self):
assert _parse_dt("2026-07-27T06:02:05") == datetime(2026, 7, 27, 6, 2, 5)
@pytest.mark.parametrize("value", ["", None, "not a date", 12345, {}])
def test_returns_none_for_junk(self, value):
assert _parse_dt(value) is None
class TestSettingKeyBlocklist:
@pytest.mark.parametrize(
"key",
[
"bambu_cloud_token",
"auth_secret_key",
"ha_token",
"prometheus_token",
"printer_access_code",
"smtp_password",
"some_api_key",
"ftp_passphrase",
"MQTT_SECRET",
],
)
def test_credential_like_keys_are_blocked(self, key):
assert _is_blocked_setting_key(key) is True
@pytest.mark.parametrize(
"key",
["low_stock_threshold", "currency", "theme", "local_backup_enabled", "timezone"],
)
def test_ordinary_keys_are_allowed(self, key):
assert _is_blocked_setting_key(key) is False
@pytest.mark.parametrize(
"key",
["auth_enabled", "advanced_auth_enabled", "local_login_enabled", "setup_completed"],
)
def test_auth_policy_keys_are_protected(self, key):
# Not credential-shaped, so the secret hints never catch them.
assert _is_blocked_setting_key(key) is False
assert _is_protected_setting_key(key) is True
@pytest.mark.parametrize("key", ["currency", "ldap_enabled", "auth_secret_key"])
def test_protected_set_is_only_the_auth_policy_keys(self, key):
assert _is_protected_setting_key(key) is False
def test_ha_token_from_env_is_deliberately_not_carved_out(self):
"""Recorded so the review's question about it is not re-litigated.
``ha_token_from_env`` looks like a false positive for the ``token`` hint,
but it is only ever constructed in the settings GET response
(``get_homeassistant_settings``). It is absent from ``AppSettingsUpdate``
and so is never a ``Settings`` row — it cannot reach a backup, which
makes an allowlist entry for it dead code.
Carving it out would also be a live hole rather than a tidy-up: an
attacker-authored ``settings/app_settings.json`` could then get a
``*token*``-named row written simply by choosing that name. This
blocklist's whole job is belt-and-braces, so a name-shaped exception to
it is exactly the wrong shape of fix.
"""
assert _is_blocked_setting_key("ha_token_from_env") is True
class TestCategoryTally:
def test_a_note_carries_code_params_and_english(self):
tally = _CategoryTally()
tally.note("noData", "No data of this kind in this backup")
tally.note("spoolUsageUnresolved", "2 usage record(s) skipped", count=2)
assert tally.notes == [
{"code": "noData", "params": {}, "message": "No data of this kind in this backup"},
{"code": "spoolUsageUnresolved", "params": {"count": 2}, "message": "2 usage record(s) skipped"},
]
def test_notes_are_deduplicated(self):
tally = _CategoryTally()
tally.note("noData", "same")
tally.note("noData", "same")
assert len(tally.notes) == 1
def test_the_same_code_with_different_params_is_kept(self):
"""Two printers can both be offline, and the user needs both names."""
tally = _CategoryTally()
tally.note("kprofilesPrinterOffline", "A is not connected", printer="A")
tally.note("kprofilesPrinterOffline", "B is not connected", printer="B")
assert len(tally.notes) == 2
def test_notes_are_bounded(self):
tally = _CategoryTally()
for i in range(50):
tally.note("noData", f"note {i}", index=i)
assert len(tally.notes) == 20
class TestRestoreRequestSchema:
def test_rejects_empty_category_list(self):
with pytest.raises(ValueError):
GitHubRestoreRequest(categories=[])
def test_deduplicates_categories(self):
request = GitHubRestoreRequest(
categories=[RestoreCategory.SPOOLS, RestoreCategory.SPOOLS, RestoreCategory.SETTINGS]
)
assert request.categories == [RestoreCategory.SPOOLS, RestoreCategory.SETTINGS]
def test_defaults_to_head(self):
assert GitHubRestoreRequest(categories=[RestoreCategory.SPOOLS]).ref == "HEAD"
@pytest.mark.parametrize("ref", ["HEAD", "abc1234", "a" * 40])
def test_accepts_valid_refs(self, ref):
assert GitHubRestoreRequest(ref=ref, categories=[RestoreCategory.SPOOLS]).ref == ref
@pytest.mark.parametrize("ref", ["abc", "main", "../etc/passwd", "a" * 41, "zzzzzzz", "abc 123"])
def test_rejects_refs_that_are_not_object_names(self, ref):
with pytest.raises(ValueError):
GitHubRestoreRequest(ref=ref, categories=[RestoreCategory.SPOOLS])
class TestRestoreSettings:
@pytest.mark.asyncio
async def test_inserts_missing_keys(self, db_session):
tally = _CategoryTally()
payload = {"version": "1.0", "settings": {"currency": "EUR", "theme": "dark"}}
await _service()._restore_settings(db_session, payload, overwrite=False, tally=tally)
await db_session.commit()
rows = {s.key: s.value for s in (await db_session.execute(select(Settings))).scalars().all()}
assert rows == {"currency": "EUR", "theme": "dark"}
assert tally.restored == 2
@pytest.mark.asyncio
async def test_skips_existing_key_when_overwrite_off(self, db_session):
db_session.add(Settings(key="currency", value="USD"))
await db_session.commit()
tally = _CategoryTally()
await _service()._restore_settings(db_session, {"settings": {"currency": "EUR"}}, overwrite=False, tally=tally)
await db_session.commit()
row = (await db_session.execute(select(Settings).where(Settings.key == "currency"))).scalar_one()
assert row.value == "USD"
assert tally.skipped == 1
assert tally.restored == 0
@pytest.mark.asyncio
async def test_overwrites_existing_key_when_enabled(self, db_session):
db_session.add(Settings(key="currency", value="USD"))
await db_session.commit()
tally = _CategoryTally()
await _service()._restore_settings(db_session, {"settings": {"currency": "EUR"}}, overwrite=True, tally=tally)
await db_session.commit()
row = (await db_session.execute(select(Settings).where(Settings.key == "currency"))).scalar_one()
assert row.value == "EUR"
assert tally.restored == 1
@pytest.mark.asyncio
async def test_credential_keys_are_never_restored(self, db_session):
"""A backup predating the collector's denylist can still contain secrets."""
tally = _CategoryTally()
payload = {"settings": {"currency": "EUR", "bambu_cloud_token": "leaked", "ha_token": "leaked"}}
await _service()._restore_settings(db_session, payload, overwrite=True, tally=tally)
await db_session.commit()
keys = {s.key for s in (await db_session.execute(select(Settings))).scalars().all()}
assert keys == {"currency"}
# Refusals are notes, not tally rows: the preview never counted these
# keys, so counting them here would put the total above what the user
# was shown before they pressed Restore.
assert tally.skipped == 0
assert any("credential-like" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_auth_settings_are_never_restored(self, db_session):
"""Restoring auth_enabled=false would disable auth behind the cache's back."""
db_session.add(Settings(key="auth_enabled", value="true"))
db_session.add(Settings(key="local_login_enabled", value="true"))
await db_session.commit()
tally = _CategoryTally()
payload = {
"settings": {
"currency": "EUR",
"auth_enabled": "false",
"advanced_auth_enabled": "false",
"local_login_enabled": "false",
"setup_completed": "false",
}
}
await _service()._restore_settings(db_session, payload, overwrite=True, tally=tally)
await db_session.commit()
rows = {s.key: s.value for s in (await db_session.execute(select(Settings))).scalars().all()}
assert rows["auth_enabled"] == "true"
assert rows["local_login_enabled"] == "true"
assert "advanced_auth_enabled" not in rows
assert "setup_completed" not in rows
assert rows["currency"] == "EUR"
assert tally.restored == 1
# As above: refused keys are outside the preview's count, so outside the
# tally too.
assert tally.skipped == 0
assert any("authentication setting" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_missing_payload_is_noted_not_fatal(self, db_session):
tally = _CategoryTally()
await _service()._restore_settings(db_session, None, overwrite=True, tally=tally)
assert tally.restored == 0
assert _codes(tally) == ["noData"]
class TestSettingValueIsTrue:
"""Only the spellings a reader actually treats as "on" count as on."""
@pytest.mark.parametrize("value", ["true", "TRUE", " True ", True])
def test_on(self, value):
assert _setting_value_is_true(value) is True
@pytest.mark.parametrize("value", ["false", "1", "on", "yes", "", None, False, 0])
def test_off(self, value):
# "1"/"on"/"yes" are deliberately off: no reader in the codebase treats
# them as on, so restoring one cannot switch anything on either.
assert _setting_value_is_true(value) is False
class TestUsableCredential:
@pytest.mark.parametrize("value", ["s3cret", " x "])
def test_present_values_are_usable(self, value):
assert _is_usable_credential(value) is True
@pytest.mark.parametrize("value", [None, "", " "])
def test_absent_or_blank_is_not(self, value):
# A present-but-blank prometheus_token row is exactly the `if token:`
# hole in the metrics route, so it must not count as protection.
assert _is_usable_credential(value) is False
class TestCompanionCredentials:
"""Toggles whose safety depends on a credential the restore refuses to write.
``prometheus_enabled`` is the sharp one. ``/api/v1/metrics`` is a public
route whose only gate is a non-empty ``prometheus_token``, so restoring the
toggle onto an instance that has no token row publishes the entire metrics
body to anyone who can reach the port — and with overwrite *off*, since the
row is missing rather than present. The other four break an integration
rather than open one, but they are the same shape.
"""
async def _restore(self, db, tally=None, overwrite=False, **settings) -> _CategoryTally:
tally = tally or _CategoryTally()
await _service()._restore_settings(db, {"settings": settings}, overwrite=overwrite, tally=tally)
await db.commit()
return tally
async def _rows(self, db) -> dict:
return {s.key: s.value for s in (await db.execute(select(Settings))).scalars().all()}
# --- The refusal itself ------------------------------------------------
@pytest.mark.asyncio
async def test_prometheus_toggle_is_refused_when_its_token_was_skipped(self, db_session):
"""The headline case: overwrite off, empty database, endpoint stays shut."""
tally = await self._restore(db_session, currency="EUR", prometheus_enabled="true", prometheus_token="s3cret")
rows = await self._rows(db_session)
assert rows == {"currency": "EUR"}
assert any("prometheus_enabled" in note and "switched off" in note for note in _messages(tally))
@pytest.mark.asyncio
@pytest.mark.parametrize("toggle,credential", sorted(_COMPANION_CREDENTIALS.items()))
async def test_every_pair_refuses_its_toggle(self, db_session, toggle, credential, monkeypatch):
monkeypatch.delenv("HA_TOKEN", raising=False)
await self._restore(db_session, **{toggle: "true", credential: "s3cret"})
assert toggle not in await self._rows(db_session)
@pytest.mark.asyncio
async def test_ha_toggle_is_refused_when_the_environment_has_no_token(self, db_session, monkeypatch):
monkeypatch.delenv("HA_TOKEN", raising=False)
await self._restore(db_session, ha_enabled="true", ha_token="s3cret", ha_url="http://ha.local")
rows = await self._rows(db_session)
assert "ha_enabled" not in rows
assert rows["ha_url"] == "http://ha.local"
@pytest.mark.asyncio
async def test_a_blank_local_credential_row_is_not_usable(self, db_session):
db_session.add(Settings(key="prometheus_token", value=""))
await db_session.commit()
await self._restore(db_session, prometheus_enabled="true", prometheus_token="s3cret")
assert "prometheus_enabled" not in await self._rows(db_session)
@pytest.mark.asyncio
@pytest.mark.parametrize("value", ["TRUE", " True ", True])
async def test_true_is_refused_however_it_is_spelled(self, db_session, value):
await self._restore(db_session, prometheus_enabled=value, prometheus_token="s3cret")
assert "prometheus_enabled" not in await self._rows(db_session)
# --- Ruling 3: the tally counts what the preview counted ---------------
@pytest.mark.asyncio
async def test_refusals_are_not_counted_in_the_tally(self, db_session):
tally = await self._restore(db_session, currency="EUR", prometheus_enabled="true", prometheus_token="s3cret")
assert (tally.restored, tally.skipped, tally.failed) == (1, 0, 0)
@pytest.mark.asyncio
async def test_tally_total_equals_the_preview_item_count(self, db_session):
"""The ruling, encoded: the user is shown a number, and it has to hold.
Off by three before this change — the two name-based refusals and the
companion one were all counted as ``skipped`` despite never being in the
preview's count.
"""
db_session.add(Settings(key="theme", value="light"))
await db_session.commit()
values = {
"currency": "EUR", # inserted -> restored
"theme": "dark", # exists, overwrite off -> skipped
"low_stock_threshold": None, # no value -> skipped
"": "junk", # unusable key -> failed
"bambu_cloud_token": "x", # blocked -> refused
"auth_enabled": "false", # protected -> refused
"prometheus_enabled": "true", # companion -> refused
"prometheus_token": "s3cret", # blocked -> refused
}
item_count, _ = await _service()._count_items(
db_session, RestoreCategory.SETTINGS, {SETTINGS_PATH: {"settings": values}}
)
tally = _CategoryTally()
await _service()._restore_settings(db_session, {"settings": values}, overwrite=False, tally=tally)
await db_session.commit()
assert tally.restored + tally.skipped + tally.failed == item_count
assert (tally.restored, tally.skipped, tally.failed) == (1, 2, 1)
@pytest.mark.asyncio
async def test_preview_count_drops_by_one_when_the_local_credential_is_missing(self, db_session):
parsed = {
SETTINGS_PATH: {"settings": {"currency": "EUR", "prometheus_enabled": "true", "prometheus_token": "s3cret"}}
}
refused_count, refused_detail = await _service()._count_items(db_session, RestoreCategory.SETTINGS, parsed)
db_session.add(Settings(key="prometheus_token", value="already-set"))
await db_session.commit()
allowed_count, allowed_detail = await _service()._count_items(db_session, RestoreCategory.SETTINGS, parsed)
assert refused_count == allowed_count - 1
assert refused_detail.code == "settingsCompanionWillSkip"
assert refused_detail.params == {"count": 1, "companion": 1}
# Nothing is being left off now, so the wording drops back to the plain
# credential caveat.
assert allowed_detail.code == "settingsCredentialsWillSkip"
# --- The exposure class: a blank backup credential is the hole ----------
#
# The rule's second condition — "the backup carried a usable credential" —
# is what stops it refusing an anonymous MQTT broker. It does not transfer to
# Prometheus: a backup taken on an instance that enabled Prometheus without a
# token (the field is optional and defaults to "") carries the toggle and no
# usable token, and writing it opens /api/v1/metrics just as wide. That is
# the *more* likely source of the exposure, not the less.
@pytest.mark.asyncio
async def test_prometheus_is_refused_when_the_backup_has_no_token_key_at_all(self, db_session):
tally = await self._restore(db_session, currency="EUR", prometheus_enabled="true")
rows = await self._rows(db_session)
assert rows == {"currency": "EUR"}
assert any("prometheus_enabled" in note and "switched off" in note for note in _messages(tally))
@pytest.mark.asyncio
@pytest.mark.parametrize("token", ["", " "])
async def test_prometheus_is_refused_when_the_backup_token_is_blank(self, db_session, token):
tally = await self._restore(db_session, prometheus_enabled="true", prometheus_token=token)
assert "prometheus_enabled" not in await self._rows(db_session)
assert "settingsCompanionSkipped" in _codes(tally)
@pytest.mark.asyncio
async def test_the_preview_says_so_with_no_credential_key_to_skip(self, db_session):
"""The wording has to survive ``blocked`` being empty.
The shared caveat counts credential-like keys *and* switches; on this
payload there are no credential-like keys, so "0 credential-like key(s)
will be skipped" would be noise.
"""
parsed = {SETTINGS_PATH: {"settings": {"currency": "EUR", "prometheus_enabled": "true"}}}
count, detail = await _service()._count_items(db_session, RestoreCategory.SETTINGS, parsed)
assert count == 1
assert detail.code == "settingsCompanionOnlyWillSkip"
assert detail.params == {"companion": 1}
@pytest.mark.asyncio
async def test_the_availability_class_keeps_the_backup_credential_condition(self, db_session):
"""The other half of the same change: only Prometheus loses condition 2.
Absent is treated like blank here — an anonymous broker or bind is a
working config, so refusing it would be a false positive.
"""
await self._restore(db_session, mqtt_enabled="true", ldap_enabled="true", virtual_printer_enabled="true")
rows = await self._rows(db_session)
assert rows["mqtt_enabled"] == "true"
assert rows["ldap_enabled"] == "true"
assert rows["virtual_printer_enabled"] == "true"
def test_every_exposure_toggle_is_a_companion_toggle(self):
assert _COMPANION_EXPOSURE_TOGGLES.issubset(_COMPANION_CREDENTIALS)
# --- Controls: over-refusal is the real risk here ----------------------
@pytest.mark.asyncio
async def test_a_usable_local_credential_lets_the_toggle_through(self, db_session):
db_session.add(Settings(key="prometheus_token", value="already-set"))
await db_session.commit()
tally = await self._restore(db_session, prometheus_enabled="true", prometheus_token="s3cret")
assert (await self._rows(db_session))["prometheus_enabled"] == "true"
assert not any("switched off" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_the_exposure_route_still_stands_down_for_a_local_token(self, db_session):
"""Skipping condition 2 must not skip the local-state pass with it."""
db_session.add(Settings(key="prometheus_token", value="already-set"))
await db_session.commit()
tally = await self._restore(db_session, prometheus_enabled="true")
assert (await self._rows(db_session))["prometheus_enabled"] == "true"
assert not any("switched off" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_the_exposure_route_still_stands_down_when_already_on(self, db_session):
"""The exposure pre-dates this restore either way — see ruling 3."""
db_session.add(Settings(key="prometheus_enabled", value="true"))
await db_session.commit()
tally = await self._restore(db_session, overwrite=True, prometheus_enabled="true")
assert (await self._rows(db_session))["prometheus_enabled"] == "true"
assert not any("switched off" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_an_anonymous_broker_is_not_a_false_positive(self, db_session):
"""mqtt_relay passes an empty password straight through — a real config."""
tally = await self._restore(db_session, mqtt_enabled="true", mqtt_broker="10.0.0.5")
assert (await self._rows(db_session))["mqtt_enabled"] == "true"
assert not any("switched off" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_an_anonymous_ldap_bind_is_not_a_false_positive(self, db_session):
"""Same for a backup that carries the key with a blank value."""
tally = await self._restore(db_session, ldap_enabled="true", ldap_bind_password=" ")
assert (await self._rows(db_session))["ldap_enabled"] == "true"
assert not any("switched off" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_turning_a_toggle_off_is_always_written(self, db_session):
await self._restore(db_session, prometheus_enabled="false", prometheus_token="s3cret")
assert (await self._rows(db_session))["prometheus_enabled"] == "false"
@pytest.mark.asyncio
@pytest.mark.parametrize("value", ["1", "on", "yes"])
async def test_spellings_no_reader_treats_as_on_are_written(self, db_session, value):
await self._restore(db_session, prometheus_enabled=value, prometheus_token="s3cret")
assert (await self._rows(db_session))["prometheus_enabled"] == value
@pytest.mark.asyncio
async def test_ha_token_in_the_environment_counts_as_usable(self, db_session, monkeypatch):
monkeypatch.setenv("HA_TOKEN", "from-env")
await self._restore(db_session, ha_enabled="true", ha_token="s3cret")
assert (await self._rows(db_session))["ha_enabled"] == "true"
@pytest.mark.asyncio
async def test_a_toggle_already_on_locally_is_written(self, db_session):
"""The exposure pre-dates the restore, so "left switched off" would be a lie."""
db_session.add(Settings(key="prometheus_enabled", value="true"))
await db_session.commit()
tally = await self._restore(db_session, overwrite=True, prometheus_enabled="true", prometheus_token="s3cret")
assert (await self._rows(db_session))["prometheus_enabled"] == "true"
assert not any("switched off" in note for note in _messages(tally))
# --- The map itself ----------------------------------------------------
def test_every_companion_credential_is_blocked_and_no_toggle_is(self):
"""Guards the rule against a future edit to _SECRET_KEY_HINTS.
If a credential stopped being blocked, its toggle would travel with it
and the refusal would be pointless; if a toggle started being blocked,
the pair would never be reached at all.
"""
for toggle, credential in _COMPANION_CREDENTIALS.items():
assert _is_blocked_setting_key(credential) is True, credential
assert _is_blocked_setting_key(toggle) is False, toggle
assert _is_protected_setting_key(toggle) is False, toggle
def test_every_environment_override_names_a_companion_credential(self):
assert set(_COMPANION_CREDENTIAL_ENV) <= set(_COMPANION_CREDENTIALS.values())
@pytest.mark.asyncio
async def test_plan_leaves_unusable_key_names_in_no_bucket(self, db_session):
"""They are the restore's ``failed``, not a refusal."""
plan = await _service()._plan_settings(db_session, {"": "x", 7: "y", "currency": "EUR"})
assert plan == _SettingsPlan()
class TestSpoolTagOverwrite:
"""Overwrite must not write the backup's *other* tag key onto a matched spool.
``tag_uid`` and ``tray_uuid`` are both in the overwrite ``setattr`` loop, and
neither column has a unique constraint, so writing one onto a spool matched
by the other silently creates a duplicate tag rather than erroring. After
that ``_find_spool``'s ``.first()`` is non-deterministic and an AMS tag
lookup resolves to an arbitrary one of the two. The same loop can also clear
a tag the user has scanned since the backup was taken.
"""
def _entry(self, **overrides):
entry = {
"id": 41,
"material": "PLA",
"brand": "Bambu Lab",
"created_at": "2026-01-05 12:00:00",
"tag_uid": "TAG-A",
"tray_uuid": None,
}
entry.update(overrides)
return entry
async def _restore(self, db, entry, tally=None):
tally = tally or _CategoryTally()
await _service()._restore_spools(db, {"spools": [entry]}, None, True, tally, {})
await db.commit()
return tally
@pytest.mark.asyncio
async def test_an_empty_incoming_tag_does_not_clear_a_scanned_one(self, db_session):
"""The backup predates the scan, so the local tag is the newer fact."""
db_session.add(Spool(material="PLA", brand="Bambu Lab", tag_uid="TAG-A", tray_uuid="TRAY-LIVE"))
await db_session.commit()
tally = await self._restore(db_session, self._entry(tray_uuid=None))
row = (await db_session.execute(select(Spool))).scalar_one()
assert row.tray_uuid == "TRAY-LIVE"
assert any(note["code"] == "spoolTagKept" for note in tally.notes)
@pytest.mark.asyncio
async def test_a_tag_another_spool_already_holds_is_not_written(self, db_session):
db_session.add(Spool(material="PLA", brand="Bambu Lab", tag_uid="TAG-A"))
db_session.add(Spool(material="PETG", brand="Other", tray_uuid="TRAY-B"))
await db_session.commit()
tally = await self._restore(db_session, self._entry(tray_uuid="TRAY-B"))
holders = (await db_session.execute(select(Spool).where(Spool.tray_uuid == "TRAY-B"))).scalars().all()
assert len(holders) == 1, "a duplicate tray_uuid makes AMS lookups non-deterministic"
assert holders[0].material == "PETG"
assert any(note["code"] == "spoolTagKept" for note in tally.notes)
@pytest.mark.asyncio
async def test_the_note_counts_every_column_it_kept(self, db_session):
db_session.add(Spool(material="PLA", brand="Bambu Lab", tag_uid="TAG-A", tray_uuid="TRAY-LIVE"))
db_session.add(Spool(material="PETG", brand="Other", tag_uid="TAG-CLASH"))
await db_session.commit()
# Matched on tray_uuid, so the guard judges tag_uid: it clashes.
tally = await self._restore(db_session, self._entry(tag_uid="TAG-CLASH", tray_uuid="TRAY-LIVE"))
row = (await db_session.execute(select(Spool).where(Spool.tray_uuid == "TRAY-LIVE"))).scalar_one()
assert row.tag_uid == "TAG-A"
note = next(n for n in tally.notes if n["code"] == "spoolTagKept")
assert note["params"] == {"count": 1}
# --- Controls ----------------------------------------------------------
@pytest.mark.asyncio
async def test_a_free_tag_is_still_written(self, db_session):
"""The point of overwrite: a spool that gained a tray_uuid gets it."""
db_session.add(Spool(material="PLA", brand="Bambu Lab", tag_uid="TAG-A"))
await db_session.commit()
tally = await self._restore(db_session, self._entry(tray_uuid="TRAY-NEW"))
row = (await db_session.execute(select(Spool))).scalar_one()
assert row.tray_uuid == "TRAY-NEW"
assert not any(note["code"] == "spoolTagKept" for note in tally.notes)
@pytest.mark.asyncio
async def test_an_unchanged_tag_is_not_reported_as_kept(self, db_session):
db_session.add(Spool(material="PLA", brand="Bambu Lab", tag_uid="TAG-A", tray_uuid="TRAY-A"))
await db_session.commit()
tally = await self._restore(db_session, self._entry(tray_uuid="TRAY-A"))
assert not any(note["code"] == "spoolTagKept" for note in tally.notes)
@pytest.mark.asyncio
async def test_a_new_spool_keeps_both_tags_from_the_backup(self, db_session):
"""The guard is an overwrite-only concern; an insert is unaffected."""
await self._restore(db_session, self._entry(tag_uid="TAG-NEW", tray_uuid="TRAY-NEW"))
row = (await db_session.execute(select(Spool))).scalar_one()
assert (row.tag_uid, row.tray_uuid) == ("TAG-NEW", "TRAY-NEW")
@pytest.mark.asyncio
async def test_find_spool_reports_which_key_matched(self, db_session):
db_session.add(Spool(material="PLA", tag_uid="TAG-A"))
db_session.add(Spool(material="PETG", tray_uuid="TRAY-B"))
await db_session.commit()
service = _service()
assert (await service._find_spool(db_session, {"tag_uid": "TAG-A"}))[1] == "tag_uid"
assert (await service._find_spool(db_session, {"tray_uuid": "TRAY-B"}))[1] == "tray_uuid"
assert await service._find_spool(db_session, {"tag_uid": "NOPE"}) == (None, None)
class TestRestoreSpools:
def _spool_entry(self, **overrides):
entry = {
"id": 41,
"material": "PLA",
"subtype": "Basic",
"color_name": "Jade White",
"brand": "Bambu Lab",
"tag_uid": "AABBCCDD",
"created_at": "2026-01-05 12:00:00",
"weight_used": 120.5,
}
entry.update(overrides)
return entry
@pytest.mark.asyncio
async def test_inserts_without_reusing_backup_id(self, db_session):
"""The backup's spool.id belongs to an unrelated row today."""
db_session.add(Spool(material="PETG")) # occupies id 1
await db_session.commit()
tally = _CategoryTally()
payload = {"spools": [self._spool_entry(id=1)]}
await _service()._restore_spools(db_session, payload, None, False, tally, {})
await db_session.commit()
spools = (await db_session.execute(select(Spool))).scalars().all()
assert len(spools) == 2
restored = next(s for s in spools if s.tag_uid == "AABBCCDD")
assert restored.id != 1
assert restored.material == "PLA"
@pytest.mark.asyncio
async def test_matches_existing_spool_by_tag_uid(self, db_session):
db_session.add(Spool(material="PLA", tag_uid="AABBCCDD", color_name="Old"))
await db_session.commit()
tally = _CategoryTally()
await _service()._restore_spools(db_session, {"spools": [self._spool_entry()]}, None, False, tally, {})
await db_session.commit()
assert len((await db_session.execute(select(Spool))).scalars().all()) == 1
assert tally.skipped == 1
@pytest.mark.asyncio
async def test_matches_existing_spool_by_tray_uuid(self, db_session):
db_session.add(Spool(material="PLA", tray_uuid="1234" * 8))
await db_session.commit()
tally = _CategoryTally()
entry = self._spool_entry(tag_uid=None, tray_uuid="1234" * 8)
await _service()._restore_spools(db_session, {"spools": [entry]}, None, False, tally, {})
await db_session.commit()
assert len((await db_session.execute(select(Spool))).scalars().all()) == 1
assert tally.skipped == 1
@pytest.mark.asyncio
async def test_matches_tagless_spool_by_descriptive_composite(self, db_session):
"""Manually added spools have no tag, so fall back to created_at + description."""
db_session.add(
Spool(
material="PLA",
subtype="Basic",
color_name="Jade White",
brand="Bambu Lab",
created_at=datetime(2026, 1, 5, 12, 0, 0),
)
)
await db_session.commit()
tally = _CategoryTally()
entry = self._spool_entry(tag_uid=None)
await _service()._restore_spools(db_session, {"spools": [entry]}, None, False, tally, {})
await db_session.commit()
assert len((await db_session.execute(select(Spool))).scalars().all()) == 1
assert tally.skipped == 1
@pytest.mark.asyncio
async def test_overwrite_updates_matched_spool(self, db_session):
db_session.add(Spool(material="PLA", tag_uid="AABBCCDD", color_name="Old", weight_used=0))
await db_session.commit()
tally = _CategoryTally()
await _service()._restore_spools(db_session, {"spools": [self._spool_entry()]}, None, True, tally, {})
await db_session.commit()
row = (await db_session.execute(select(Spool))).scalar_one()
assert row.color_name == "Jade White"
assert row.weight_used == 120.5
assert tally.restored == 1
@pytest.mark.asyncio
async def test_insert_preserves_created_at_so_repeat_restore_is_idempotent(self, db_session):
"""Second restore of the same backup must match, not duplicate."""
service = _service()
payload = {"spools": [self._spool_entry(tag_uid=None)]}
await service._restore_spools(db_session, payload, None, False, _CategoryTally(), {})
await db_session.commit()
await service._restore_spools(db_session, payload, None, False, _CategoryTally(), {})
await db_session.commit()
spools = (await db_session.execute(select(Spool))).scalars().all()
assert len(spools) == 1
assert spools[0].created_at == datetime(2026, 1, 5, 12, 0, 0)
@pytest.mark.asyncio
async def test_usage_history_spool_id_is_remapped(self, db_session):
"""Usage rows must point at the new local spool id, not the backup's."""
tally = _CategoryTally()
inventory = {"spools": [self._spool_entry(id=41)]}
usage = {
"usage_history": [
{
"id": 900,
"spool_id": 41,
"printer_id": None,
"print_name": "benchy.3mf",
"archive_id": None,
"weight_used": 12.0,
"percent_used": 5,
"status": "completed",
"created_at": "2026-02-01 09:00:00",
}
]
}
await _service()._restore_spools(db_session, inventory, usage, False, tally, {})
await db_session.commit()
spool = (await db_session.execute(select(Spool))).scalar_one()
row = (await db_session.execute(select(SpoolUsageHistory))).scalar_one()
assert row.spool_id == spool.id
assert row.print_name == "benchy.3mf"
@pytest.mark.asyncio
async def test_usage_history_archive_id_is_remapped(self, db_session):
tally = _CategoryTally()
inventory = {"spools": [self._spool_entry(id=41)]}
usage = {
"usage_history": [
{
"spool_id": 41,
"archive_id": 77,
"weight_used": 1.0,
"created_at": "2026-02-01 09:00:00",
}
]
}
archive = PrintArchive(filename="a.3mf", file_path="", file_size=1)
db_session.add(archive)
await db_session.flush()
await _service()._restore_spools(db_session, inventory, usage, False, tally, {77: archive.id})
await db_session.commit()
row = (await db_session.execute(select(SpoolUsageHistory))).scalar_one()
assert row.archive_id == archive.id
@pytest.mark.asyncio
async def test_usage_row_with_unresolvable_spool_is_skipped_and_explained(self, db_session):
tally = _CategoryTally()
usage = {"usage_history": [{"spool_id": 999, "weight_used": 1.0, "created_at": "2026-02-01 09:00:00"}]}
await _service()._restore_spools(db_session, {"spools": []}, usage, False, tally, {})
await db_session.commit()
assert (await db_session.execute(select(SpoolUsageHistory))).scalars().first() is None
assert tally.skipped == 1
assert any("their spool is not in this backup's spool list" in note for note in _messages(tally))
# No remedy is offered, because none exists: overwrite does not change
# which spools land in the map (a skipped spool is mapped anyway), and
# usage history is always restored alongside the spools category.
assert not any("overwrite" in note.lower() for note in _messages(tally))
@pytest.mark.asyncio
async def test_usage_resolves_against_a_spool_skipped_because_overwrite_is_off(self, db_session):
"""A skipped spool is still mapped, so its usage rows are not "unresolved".
This is why the note above offers no remedy: turning overwrite on would
not rescue anything, and saying so misdescribed which records are lost.
"""
db_session.add(Spool(material="PLA", tag_uid="AABBCCDD", color_name="Old"))
await db_session.commit()
tally = _CategoryTally()
inventory = {"spools": [self._spool_entry(id=41)]}
usage = {
"usage_history": [
{"spool_id": 41, "print_name": "b.3mf", "weight_used": 5.0, "created_at": "2026-02-01 09:00:00"}
]
}
await _service()._restore_spools(db_session, inventory, usage, False, tally, {})
await db_session.commit()
spool = (await db_session.execute(select(Spool))).scalar_one()
row = (await db_session.execute(select(SpoolUsageHistory))).scalar_one()
assert row.spool_id == spool.id
assert not any("spool list" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_usage_history_is_not_duplicated_on_repeat_restore(self, db_session):
service = _service()
inventory = {"spools": [self._spool_entry(id=41)]}
usage = {
"usage_history": [
{"spool_id": 41, "print_name": "b.3mf", "weight_used": 5.0, "created_at": "2026-02-01 09:00:00"}
]
}
await service._restore_spools(db_session, inventory, usage, False, _CategoryTally(), {})
await db_session.commit()
await service._restore_spools(db_session, inventory, usage, False, _CategoryTally(), {})
await db_session.commit()
rows = (await db_session.execute(select(SpoolUsageHistory))).scalars().all()
assert len(rows) == 1
@pytest.mark.asyncio
async def test_dropped_archive_link_is_explained(self, db_session):
"""Spools without archives nulls every usage -> archive link, silently."""
tally = _CategoryTally()
inventory = {"spools": [self._spool_entry(id=41)]}
usage = {
"usage_history": [
{"spool_id": 41, "archive_id": 7, "weight_used": 1.0, "created_at": "2026-02-01 09:00:00"},
{"spool_id": 41, "archive_id": 8, "weight_used": 2.0, "created_at": "2026-02-01 10:00:00"},
{"spool_id": 41, "weight_used": 3.0, "created_at": "2026-02-01 11:00:00"},
]
}
# Empty archive_id_map: the archives category wasn't selected, so its
# payload was never fetched and there is nothing to match against.
await _service()._restore_spools(db_session, inventory, usage, False, tally, {})
await db_session.commit()
rows = (await db_session.execute(select(SpoolUsageHistory))).scalars().all()
assert len(rows) == 3
assert all(row.archive_id is None for row in rows)
# Only the two that had a link to lose are counted.
assert any("2 usage record(s) restored without their print-history link" in n for n in _messages(tally))
assert any("select Print archives alongside" in n for n in _messages(tally))
@pytest.mark.asyncio
async def test_no_note_when_every_archive_link_resolves(self, db_session):
tally = _CategoryTally()
inventory = {"spools": [self._spool_entry(id=41)]}
usage = {
"usage_history": [
{"spool_id": 41, "archive_id": 7, "weight_used": 1.0, "created_at": "2026-02-01 09:00:00"}
]
}
archive = PrintArchive(filename="linked.3mf", file_path="", file_size=1)
db_session.add(archive)
await db_session.flush()
await _service()._restore_spools(db_session, inventory, usage, False, tally, {7: archive.id})
await db_session.commit()
row = (await db_session.execute(select(SpoolUsageHistory))).scalar_one()
assert row.archive_id == archive.id
assert not any("print-history link" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_dangling_printer_id_is_cleared(self, db_session):
tally = _CategoryTally()
inventory = {"spools": [self._spool_entry(id=41)]}
usage = {
"usage_history": [
{"spool_id": 41, "printer_id": 4242, "weight_used": 1.0, "created_at": "2026-02-01 09:00:00"}
]
}
await _service()._restore_spools(db_session, inventory, usage, False, tally, {})
await db_session.commit()
row = (await db_session.execute(select(SpoolUsageHistory))).scalar_one()
assert row.printer_id is None
class TestRestoreArchives:
def _archive_entry(self, **overrides):
entry = {
"id": 77,
"filename": "benchy.3mf",
"file_size": 2048,
"content_hash": "abc123",
"print_name": "Benchy",
"status": "completed",
"started_at": "2026-03-01 10:00:00",
"completed_at": "2026-03-01 11:00:00",
"created_at": "2026-03-01 10:00:00",
"quantity": 1,
"is_favorite": False,
}
entry.update(overrides)
return entry
@pytest.mark.asyncio
async def test_inserts_metadata_only_row_with_empty_file_path(self, db_session):
"""print_archives.file_path is NOT NULL but is not in the backup."""
tally = _CategoryTally()
id_map: dict[int, int] = {}
await _service()._restore_archives(db_session, {"archives": [self._archive_entry()]}, False, tally, id_map)
await db_session.commit()
row = (await db_session.execute(select(PrintArchive))).scalar_one()
assert row.file_path == ""
assert row.filename == "benchy.3mf"
assert row.id != 77
assert id_map == {77: row.id}
assert any("metadata only" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_matches_existing_archive_by_hash_and_start(self, db_session):
db_session.add(
PrintArchive(
filename="benchy.3mf",
file_path="/data/benchy.3mf",
file_size=2048,
content_hash="abc123",
started_at=datetime(2026, 3, 1, 10, 0, 0),
)
)
await db_session.commit()
tally = _CategoryTally()
await _service()._restore_archives(db_session, {"archives": [self._archive_entry()]}, False, tally, {})
await db_session.commit()
rows = (await db_session.execute(select(PrintArchive))).scalars().all()
assert len(rows) == 1
assert rows[0].file_path == "/data/benchy.3mf"
assert tally.skipped == 1
@pytest.mark.asyncio
async def test_falls_back_to_filename_and_start_without_hash(self, db_session):
db_session.add(
PrintArchive(
filename="benchy.3mf",
file_path="/data/benchy.3mf",
file_size=2048,
started_at=datetime(2026, 3, 1, 10, 0, 0),
)
)
await db_session.commit()
tally = _CategoryTally()
entry = self._archive_entry(content_hash=None)
await _service()._restore_archives(db_session, {"archives": [entry]}, False, tally, {})
await db_session.commit()
assert len((await db_session.execute(select(PrintArchive))).scalars().all()) == 1
assert tally.skipped == 1
@pytest.mark.asyncio
async def test_matches_archive_with_no_started_at_by_hash(self, db_session):
"""started_at is NULL for re-sliced archives, so it cannot be required.
Gating both match branches on it meant these rows never matched: every
restore re-inserted them and overwrite mode could never update them.
"""
db_session.add(
PrintArchive(
filename="benchy.3mf",
file_path="/data/benchy.3mf",
file_size=2048,
content_hash="abc123",
started_at=None,
)
)
await db_session.commit()
tally = _CategoryTally()
entry = self._archive_entry(started_at=None)
await _service()._restore_archives(db_session, {"archives": [entry]}, False, tally, {})
await db_session.commit()
assert len((await db_session.execute(select(PrintArchive))).scalars().all()) == 1
assert tally.skipped == 1
@pytest.mark.asyncio
async def test_started_at_still_discriminates_when_present(self, db_session):
"""A NULL-tolerant match must not collapse rows that do differ."""
db_session.add(
PrintArchive(
filename="benchy.3mf",
file_path="/data/benchy.3mf",
file_size=2048,
content_hash="abc123",
started_at=datetime(2026, 3, 1, 10, 0, 0),
)
)
await db_session.commit()
tally = _CategoryTally()
# Same file, no start time recorded — a different row, not that one.
entry = self._archive_entry(started_at=None)
await _service()._restore_archives(db_session, {"archives": [entry]}, False, tally, {})
await db_session.commit()
assert len((await db_session.execute(select(PrintArchive))).scalars().all()) == 2
assert tally.restored == 1
@pytest.mark.asyncio
async def test_soft_deleted_archive_is_not_restored_as_visible(self, db_session):
"""A backup keeps soft-deleted rows, so the flag has to survive.
Their row is retained on purpose (stats keep counting the filament and
energy), so without carrying deleted_at a restore turns an archive the
user deleted back into a visible one.
"""
tally = _CategoryTally()
entry = self._archive_entry(deleted_at="2026-03-02 08:00:00")
await _service()._restore_archives(db_session, {"archives": [entry]}, False, tally, {})
await db_session.commit()
row = (await db_session.execute(select(PrintArchive))).scalar_one()
assert row.deleted_at == datetime(2026, 3, 2, 8, 0, 0)
assert tally.restored == 1
@pytest.mark.asyncio
async def test_locally_deleted_archive_stays_deleted_without_overwrite(self, db_session):
db_session.add(
PrintArchive(
filename="benchy.3mf",
file_path="",
file_size=2048,
content_hash="abc123",
started_at=datetime(2026, 3, 1, 10, 0, 0),
deleted_at=datetime(2026, 3, 5, 9, 0, 0),
)
)
await db_session.commit()
tally = _CategoryTally()
# The backup predates the deletion, so its copy is live.
await _service()._restore_archives(db_session, {"archives": [self._archive_entry()]}, False, tally, {})
await db_session.commit()
row = (await db_session.execute(select(PrintArchive))).scalar_one()
assert row.deleted_at == datetime(2026, 3, 5, 9, 0, 0)
assert tally.skipped == 1
@pytest.mark.asyncio
async def test_overwrite_undeletes_a_locally_deleted_archive_and_says_so(self, db_session):
"""The entry has to *say* the archive was live — absent no longer means null.
A commit taken before the collector wrote ``deleted_at`` carries no
opinion about it, and overwrite now leaves the column alone in that
case; see ``TestRestoredArchiveOwnership``.
"""
db_session.add(
PrintArchive(
filename="benchy.3mf",
file_path="",
file_size=2048,
content_hash="abc123",
started_at=datetime(2026, 3, 1, 10, 0, 0),
deleted_at=datetime(2026, 3, 5, 9, 0, 0),
)
)
await db_session.commit()
tally = _CategoryTally()
await _service()._restore_archives(
db_session, {"archives": [self._archive_entry(deleted_at=None)]}, True, tally, {}
)
await db_session.commit()
row = (await db_session.execute(select(PrintArchive))).scalar_one()
assert row.deleted_at is None
assert tally.restored == 1
assert any("visible again" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_overwrite_updates_metadata_but_keeps_local_file_path(self, db_session):
db_session.add(
PrintArchive(
filename="benchy.3mf",
file_path="/data/benchy.3mf",
file_size=2048,
content_hash="abc123",
started_at=datetime(2026, 3, 1, 10, 0, 0),
notes="old",
)
)
await db_session.commit()
tally = _CategoryTally()
entry = self._archive_entry(notes="restored note")
await _service()._restore_archives(db_session, {"archives": [entry]}, True, tally, {})
await db_session.commit()
row = (await db_session.execute(select(PrintArchive))).scalar_one()
assert row.notes == "restored note"
# The 3MF on disk must not be orphaned by a metadata restore.
assert row.file_path == "/data/benchy.3mf"
assert tally.restored == 1
@pytest.mark.asyncio
async def test_dangling_printer_and_project_links_are_cleared(self, db_session):
tally = _CategoryTally()
entry = self._archive_entry(printer_id=4242, project_id=4343)
await _service()._restore_archives(db_session, {"archives": [entry]}, False, tally, {})
await db_session.commit()
row = (await db_session.execute(select(PrintArchive))).scalar_one()
assert row.printer_id is None
assert row.project_id is None
assert any("no longer exist" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_valid_printer_link_is_preserved(self, db_session, printer_factory):
printer = await printer_factory()
tally = _CategoryTally()
entry = self._archive_entry(printer_id=printer.id)
await _service()._restore_archives(db_session, {"archives": [entry]}, False, tally, {})
await db_session.commit()
row = (await db_session.execute(select(PrintArchive))).scalar_one()
assert row.printer_id == printer.id
@pytest.mark.asyncio
async def test_non_dict_entry_counts_as_failed(self, db_session):
tally = _CategoryTally()
await _service()._restore_archives(db_session, {"archives": ["nonsense"]}, False, tally, {})
assert tally.failed == 1
class TestRestoreKprofiles:
@staticmethod
def _live(slot_id, filament_id="GFA00", name="Bambu PLA", setting_id="PFUS123", extruder_id=0):
"""One profile as the printer currently reports it.
``extruder_id`` mirrors ``KProfile`` (bambu_mqtt.py), which has carried
it all along; single-nozzle printers report 0.
"""
return SimpleNamespace(
slot_id=slot_id,
filament_id=filament_id,
name=name,
setting_id=setting_id,
extruder_id=extruder_id,
)
def _client(self, live=None, sent="7", ack=(True, "")):
"""A connected printer client.
``set_kprofiles_batch`` returns the sequence_id it published under, not
a success flag (#2718), and the verdict arrives separately from
``await_cali_ack`` as ``(ok, detail)``.
"""
client = MagicMock()
client.state.connected = True
client.set_kprofiles_batch = MagicMock(return_value=sent)
client.await_cali_ack = AsyncMock(return_value=ack)
client.get_kprofiles = AsyncMock(return_value=list(live or []))
return client
def _payload(self, serial="00M09A123456789", nozzle="0.4"):
return {
f"kprofiles/{serial}/{nozzle}.json": {
"version": "1.0",
"printer_serial": serial,
"nozzle_diameter": nozzle,
"profiles": [
{
"slot_id": 0,
"name": "Bambu PLA",
"k_value": "0.020000",
"filament_id": "GFA00",
"nozzle_id": "HS00-0.4",
"extruder_id": 0,
"setting_id": "PFUS123",
}
],
}
}
@pytest.mark.asyncio
async def test_sends_batch_to_connected_printer(self, db_session, printer_factory):
printer = await printer_factory(serial_number="00M09A123456789")
client = self._client()
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, self._payload(), tally)
client.set_kprofiles_batch.assert_called_once()
profiles, nozzle = client.set_kprofiles_batch.call_args.args
assert nozzle == "0.4"
assert profiles[0]["name"] == "Bambu PLA"
assert profiles[0]["filament_id"] == "GFA00"
assert tally.restored == 1
assert manager.get_client.call_args.args == (printer.id,)
@pytest.mark.asyncio
async def test_always_warns_to_verify_on_the_printer(self, db_session, printer_factory):
await printer_factory(serial_number="00M09A123456789")
client = self._client()
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, self._payload(), tally)
# A refusal is now read and counted failed, so the caveat is narrowed to
# what is genuinely left uncertain: a printer that never answers.
assert any("verify the profiles on the printer" in note for note in _messages(tally))
assert any("does not answer still counts as restored" in note for note in _messages(tally))
assert not any("without acknowledgement" in note for note in _messages(tally))
assert any("always overwrite" in note for note in _messages(tally))
# --- cali_idx is resolved live, never taken from the backup -------------
#
# Regression cover for the silent no-op found testing on an X1E: the backup
# stored cali_idx 8151, a Bambuddy edit re-keyed the profile to 4606, and
# the restore aimed extrusion_cali_set at 8151. The printer dropped it and
# the tally still said "1 restored".
@pytest.mark.asyncio
async def test_uses_the_live_cali_idx_not_the_backed_up_slot(self, db_session, printer_factory):
await printer_factory(serial_number="00M09A123456789")
payload = self._payload()
payload["kprofiles/00M09A123456789/0.4.json"]["profiles"][0]["slot_id"] = 8151
client = self._client(live=[self._live(slot_id=4606)])
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, payload, tally)
client.get_kprofiles.assert_awaited_once_with(nozzle_diameter="0.4")
profiles, _ = client.set_kprofiles_batch.call_args.args
assert profiles[0]["cali_idx"] == 4606, "must address the slot that exists now"
assert profiles[0]["cali_idx"] != 8151, "must not reuse the backup's cali_idx"
assert tally.restored == 1
@pytest.mark.asyncio
async def test_matches_on_name_when_setting_id_was_regenerated(self, db_session, printer_factory):
# A delete-then-add edit mints a fresh setting_id, so the name carries
# the match instead.
await printer_factory(serial_number="00M09A123456789")
client = self._client(live=[self._live(slot_id=4606, setting_id="PF9999999999")])
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, self._payload(), tally)
profiles, _ = client.set_kprofiles_batch.call_args.args
assert profiles[0]["cali_idx"] == 4606
# The live setting_id wins: it is what the printer associates with the slot.
assert profiles[0]["setting_id"] == "PF9999999999"
@pytest.mark.asyncio
async def test_unmatched_profile_is_added_rather_than_aimed_at_a_dead_slot(self, db_session, printer_factory):
await printer_factory(serial_number="00M09A123456789")
client = self._client(live=[]) # printer has nothing for this nozzle
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, self._payload(), tally)
profiles, _ = client.set_kprofiles_batch.call_args.args
assert profiles[0]["cali_idx"] == -1, "-1 tells the printer to add a new profile"
assert profiles[0]["setting_id"] == "PFUS123", "falls back to the backed-up preset"
assert any("added as new profiles" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_different_filament_is_not_treated_as_a_match(self, db_session, printer_factory):
# Same slot, different filament — matching on slot alone would clobber
# an unrelated profile.
await printer_factory(serial_number="00M09A123456789")
client = self._client(live=[self._live(slot_id=4606, filament_id="GFB99", name="Bambu PLA")])
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, self._payload(), tally)
profiles, _ = client.set_kprofiles_batch.call_args.args
assert profiles[0]["cali_idx"] == -1
@pytest.mark.asyncio
async def test_unreadable_live_index_degrades_to_adding(self, db_session, printer_factory):
# A failed read must not abort the restore.
await printer_factory(serial_number="00M09A123456789")
client = self._client()
client.get_kprofiles = AsyncMock(side_effect=RuntimeError("mqtt timeout"))
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, self._payload(), tally)
profiles, _ = client.set_kprofiles_batch.call_args.args
assert profiles[0]["cali_idx"] == -1
assert tally.restored == 1
@pytest.mark.asyncio
async def test_sole_profile_for_a_filament_matches_without_setting_id_or_name(self, db_session, printer_factory):
await printer_factory(serial_number="00M09A123456789")
payload = self._payload()
entry = payload["kprofiles/00M09A123456789/0.4.json"]["profiles"][0]
entry["setting_id"] = None
entry["name"] = ""
client = self._client(live=[self._live(slot_id=4606, setting_id="PFOTHER", name="Renamed")])
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, payload, tally)
profiles, _ = client.set_kprofiles_batch.call_args.args
assert profiles[0]["cali_idx"] == 4606
@pytest.mark.asyncio
async def test_ambiguous_filament_without_discriminator_is_added_not_guessed(self, db_session, printer_factory):
await printer_factory(serial_number="00M09A123456789")
payload = self._payload()
entry = payload["kprofiles/00M09A123456789/0.4.json"]["profiles"][0]
entry["setting_id"] = None
entry["name"] = ""
client = self._client(live=[self._live(slot_id=1, setting_id="A"), self._live(slot_id=2, setting_id="B")])
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, payload, tally)
profiles, _ = client.set_kprofiles_batch.call_args.args
assert profiles[0]["cali_idx"] == -1, "two candidates and nothing to tell them apart"
@pytest.mark.asyncio
async def test_two_entries_cannot_claim_the_same_live_slot(self, db_session, printer_factory):
"""One live profile cannot stand in for two backed-up ones (#2656).
Both entries fell through to the single-candidate arm, both took
cali_idx 4606, both went into the batch — so the second overwrote the
first on the printer while the tally counted two restored. Reachable
whenever the user has deleted one of a pair since the backup, because
the delete-then-add re-key is what strips the setting_id match.
"""
await printer_factory(serial_number="00M09A123456789")
payload = self._payload()
entries = payload["kprofiles/00M09A123456789/0.4.json"]["profiles"]
entries[0].update(setting_id="PFGONE1", name="PLA Basic")
entries.append({**entries[0], "setting_id": "PFGONE2", "name": "PLA Matte"})
client = self._client(live=[self._live(slot_id=4606, setting_id="PFUS123", name="Bambu PLA")])
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, payload, tally)
profiles, _ = client.set_kprofiles_batch.call_args.args
assert [p["cali_idx"] for p in profiles] == [4606, -1], "the displaced entry has to be added, not aliased"
assert sum(1 for p in profiles if p["cali_idx"] == 4606) == 1
assert any("added as new profiles" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_the_displaced_entry_does_not_inherit_the_claimed_setting_id(self, db_session, printer_factory):
"""An add-as-new keeps its own preset, or it lands on top of the match anyway.
cali_idx -1 is only safe if the rest of the payload doesn't point at the
profile the first entry just claimed — the generated-setting_id fallback
reads setting_id when cali_idx is -1.
"""
await printer_factory(serial_number="00M09A123456789")
payload = self._payload()
entries = payload["kprofiles/00M09A123456789/0.4.json"]["profiles"]
entries[0].update(setting_id="PFGONE1", name="PLA Basic")
entries.append({**entries[0], "setting_id": "PFGONE2", "name": "PLA Matte"})
client = self._client(live=[self._live(slot_id=4606, setting_id="PFUS123", name="Bambu PLA")])
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, payload, _CategoryTally())
profiles, _ = client.set_kprofiles_batch.call_args.args
assert profiles[0]["setting_id"] == "PFUS123", "the match prefers the live preset"
assert profiles[1]["setting_id"] == "PFGONE2", "the displaced entry keeps its own"
@pytest.mark.asyncio
async def test_two_entries_matching_two_live_profiles_keep_their_own_slots(self, db_session, printer_factory):
"""Control: the guard must not displace a legitimate second match."""
await printer_factory(serial_number="00M09A123456789")
payload = self._payload()
entries = payload["kprofiles/00M09A123456789/0.4.json"]["profiles"]
entries.append({**entries[0], "setting_id": "PFUS456", "name": "Bambu PETG"})
client = self._client(
live=[
self._live(slot_id=4606, setting_id="PFUS123", name="Bambu PLA"),
self._live(slot_id=4607, setting_id="PFUS456", name="Bambu PETG"),
]
)
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, payload, tally)
profiles, _ = client.set_kprofiles_batch.call_args.args
assert [p["cali_idx"] for p in profiles] == [4606, 4607]
assert not any("added as new profiles" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_a_claimed_slot_does_not_make_an_ambiguous_pair_matchable(self, db_session, printer_factory):
"""Two live profiles for one filament stay ambiguous after one is taken.
The single-candidate fallback is judged against every candidate, not the
unclaimed ones — otherwise claiming the first would leave exactly one
"available" and turn a guess the code deliberately refuses into a match.
"""
await printer_factory(serial_number="00M09A123456789")
payload = self._payload()
entries = payload["kprofiles/00M09A123456789/0.4.json"]["profiles"]
entries[0].update(setting_id="PFUS123", name="Bambu PLA")
entries.append({**entries[0], "setting_id": None, "name": ""})
client = self._client(
live=[
self._live(slot_id=1, setting_id="PFUS123", name="Bambu PLA"),
self._live(slot_id=2, setting_id="PFOTHER", name="Renamed"),
]
)
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, payload, _CategoryTally())
profiles, _ = client.set_kprofiles_batch.call_args.args
assert [p["cali_idx"] for p in profiles] == [1, -1]
# --- the match is scoped to the extruder it was calibrated on -----------
#
# get_kprofiles reads per nozzle *diameter*, so on a dual-nozzle printer
# both extruders come back in one list. Scoping candidates on filament_id
# alone let one extruder's calibration be written over the other's.
@pytest.mark.asyncio
async def test_each_extruders_profile_lands_on_its_own_extruder(self, db_session, printer_factory):
"""The same preset calibrated on both extruders of an H2D.
Both live profiles share a filament_id *and* a setting_id, so the
setting_id arm matched whichever the printer happened to list first —
and with an entry per extruder the two swapped slots, each overwriting
the other's calibration while the tally counted both restored.
"""
await printer_factory(serial_number="00M09A123456789")
payload = self._payload()
entries = payload["kprofiles/00M09A123456789/0.4.json"]["profiles"]
entries.append({**entries[0], "extruder_id": 1, "nozzle_id": "HS00-0.4-R"})
client = self._client(
live=[
# Right extruder first, which is what made the bug bite.
self._live(slot_id=1001, extruder_id=1),
self._live(slot_id=1000, extruder_id=0),
]
)
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, payload, tally)
profiles, _ = client.set_kprofiles_batch.call_args.args
assert [(p["extruder_id"], p["cali_idx"]) for p in profiles] == [(0, 1000), (1, 1001)]
assert tally.restored == 2
assert not any("added as new profiles" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_the_other_extruders_profile_is_not_a_candidate(self, db_session, printer_factory):
"""One backed-up entry, and the only live profile is the other extruder's.
Adding as new is the right answer: extruder 0's calibration is not a
stand-in for extruder 1's, however well the filament and preset line up.
"""
await printer_factory(serial_number="00M09A123456789")
client = self._client(live=[self._live(slot_id=1001, extruder_id=1)])
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, self._payload(), tally)
profiles, _ = client.set_kprofiles_batch.call_args.args
assert profiles[0]["cali_idx"] == -1
assert any("added as new profiles" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_an_entry_without_an_extruder_id_still_matches(self, db_session, printer_factory):
"""Control: a pre-#2656 backup carries no extruder_id.
A missing key must leave the match exactly as it was, not turn every
entry into an add.
"""
await printer_factory(serial_number="00M09A123456789")
payload = self._payload()
payload["kprofiles/00M09A123456789/0.4.json"]["profiles"][0].pop("extruder_id")
client = self._client(live=[self._live(slot_id=4606)])
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, payload, tally)
profiles, _ = client.set_kprofiles_batch.call_args.args
assert profiles[0]["cali_idx"] == 4606
assert tally.restored == 1
@pytest.mark.asyncio
async def test_a_live_index_that_reports_no_extruder_still_matches(self, db_session, printer_factory):
"""Control: the same, for a printer whose profiles carry no extruder_id."""
await printer_factory(serial_number="00M09A123456789")
live = SimpleNamespace(slot_id=4606, filament_id="GFA00", name="Bambu PLA", setting_id="PFUS123")
client = self._client(live=[live])
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, self._payload(), tally)
profiles, _ = client.set_kprofiles_batch.call_args.args
assert profiles[0]["cali_idx"] == 4606
assert tally.restored == 1
@pytest.mark.asyncio
async def test_unknown_serial_is_skipped_with_reason(self, db_session):
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager"):
await _service()._restore_kprofiles(db_session, self._payload(serial="NOSUCH"), tally)
assert tally.restored == 0
assert tally.skipped == 1
assert any("No printer with serial NOSUCH" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_offline_printer_is_skipped_not_failed(self, db_session, printer_factory):
await printer_factory(serial_number="00M09A123456789", name="Shelf Printer")
client = MagicMock()
client.state.connected = False
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, self._payload(), tally)
assert tally.skipped == 1
assert tally.failed == 0
assert any("not connected" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_no_client_at_all_is_skipped(self, db_session, printer_factory):
await printer_factory(serial_number="00M09A123456789")
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=None)
await _service()._restore_kprofiles(db_session, self._payload(), tally)
assert tally.skipped == 1
@pytest.mark.asyncio
async def test_publish_failure_counts_as_failed(self, db_session, printer_factory):
# None is what set_kprofiles_batch returns when it could not publish —
# a disconnected client. There is no ack to wait for in that case.
await printer_factory(serial_number="00M09A123456789")
client = self._client(sent=None)
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, self._payload(), tally)
assert tally.failed == 1
assert tally.restored == 0
assert "kprofilesSendFailed" in _codes(tally)
assert "kprofilesRefused" not in _codes(tally), "nothing was sent, so the printer refused nothing"
client.await_cali_ack.assert_not_awaited()
@pytest.mark.asyncio
async def test_publish_exception_is_contained(self, db_session, printer_factory):
await printer_factory(serial_number="00M09A123456789")
client = MagicMock()
client.state.connected = True
client.set_kprofiles_batch = MagicMock(side_effect=RuntimeError("mqtt down"))
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, self._payload(), tally)
assert tally.failed == 1
# --- the printer's verdict decides the tally, not the publish ------------
#
# #2718 changed set_kprofiles_batch from returning a bool to returning the
# sequence_id it published under. A sequence_id string is truthy, so a
# restore that branches on the return value alone reports every refused
# write as saved — the defect that fix closed in every other caller.
@pytest.mark.asyncio
async def test_awaits_the_ack_for_the_sequence_id_it_was_given(self, db_session, printer_factory):
await printer_factory(serial_number="00M09A123456789")
client = self._client(sent="4211")
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, self._payload(), tally)
client.await_cali_ack.assert_awaited_once_with("4211")
assert tally.restored == 1
@pytest.mark.asyncio
async def test_a_refused_batch_counts_failed_not_restored(self, db_session, printer_factory):
await printer_factory(serial_number="00M09A123456789", name="Shelf Printer")
client = self._client(ack=(False, "invalid tray_id"))
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, self._payload(), tally)
assert tally.restored == 0
assert tally.failed == 1
assert "kprofilesRefused" in _codes(tally)
assert "kprofilesSendFailed" not in _codes(tally), "it was sent — the printer answered no"
note = next(n for n in tally.notes if n["code"] == "kprofilesRefused")
assert note["params"]["reason"] == "invalid tray_id", "the printer's own reason has to survive"
assert "Shelf Printer" in note["message"] and "invalid tray_id" in note["message"]
@pytest.mark.asyncio
async def test_a_silent_printer_still_counts_restored(self, db_session, printer_factory):
# maziggy's rule, and await_cali_ack's own contract: no answer is not
# evidence of refusal. Firmware that predates the ack never answers.
await printer_factory(serial_number="00M09A123456789")
client = self._client(ack=(True, "no acknowledgement from printer"))
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, self._payload(), tally)
assert tally.restored == 1
assert tally.failed == 0
assert "kprofilesRefused" not in _codes(tally)
@pytest.mark.asyncio
async def test_an_unreadable_ack_does_not_fail_the_batch(self, db_session, printer_factory):
# Same situation one layer up: the write most likely landed, so this
# degrades the way a timeout does rather than inventing a failure.
await printer_factory(serial_number="00M09A123456789")
client = self._client()
client.await_cali_ack = AsyncMock(side_effect=RuntimeError("mqtt down"))
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, self._payload(), tally)
assert tally.restored == 1
assert tally.failed == 0
@pytest.mark.asyncio
async def test_one_refused_nozzle_does_not_condemn_the_other(self, db_session, printer_factory):
await printer_factory(serial_number="00M09A123456789")
payload = {**self._payload(nozzle="0.4"), **self._payload(nozzle="0.8")}
client = self._client()
client.await_cali_ack = AsyncMock(side_effect=[(False, "busy"), (True, "")])
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, payload, tally)
assert tally.restored == 1
assert tally.failed == 1
@pytest.mark.asyncio
async def test_each_nozzle_is_sent_separately(self, db_session, printer_factory):
await printer_factory(serial_number="00M09A123456789")
payload = {**self._payload(nozzle="0.4"), **self._payload(nozzle="0.8")}
client = self._client()
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, payload, tally)
assert client.set_kprofiles_batch.call_count == 2
assert {c.args[1] for c in client.set_kprofiles_batch.call_args_list} == {"0.4", "0.8"}
assert tally.restored == 2
@pytest.mark.asyncio
async def test_empty_payload_is_noted(self, db_session):
tally = _CategoryTally()
await _service()._restore_kprofiles(db_session, {}, tally)
assert _codes(tally) == ["noData"]
class TestSoftDeletedArchiveRoundTrip:
"""The two halves of the soft-delete fix only work together.
The collector keeps soft-deleted rows on purpose (their stats still count),
so if it doesn't write ``deleted_at`` there is nothing for the restore to
carry across and a deleted archive comes back visible. Covered end to end
because each half looks harmless on its own.
"""
@pytest.mark.asyncio
async def test_deleted_at_survives_collect_then_restore(self, db_session):
from backend.app.services.github_backup import github_backup_service
deleted_at = datetime(2026, 3, 5, 9, 0, 0)
db_session.add(
PrintArchive(
filename="trashed.3mf",
file_path="",
file_size=1024,
content_hash="hash-trashed",
started_at=datetime(2026, 3, 1, 10, 0, 0),
deleted_at=deleted_at,
)
)
await db_session.commit()
files: dict = {}
await github_backup_service._collect_archives(db_session, files)
payload = files[ARCHIVES_PATH]
assert payload["archives"][0]["deleted_at"] == str(deleted_at)
# Restore that payload into an instance where the row is gone entirely.
await db_session.execute(PrintArchive.__table__.delete())
await db_session.commit()
tally = _CategoryTally()
await _service()._restore_archives(db_session, payload, False, tally, {})
await db_session.commit()
row = (await db_session.execute(select(PrintArchive))).scalar_one()
assert row.deleted_at == deleted_at, "a deleted archive must not come back visible"
class TestRestoredArchiveOwnership:
"""A restored archive without an owner is invisible to the person who owns it.
``created_by_id`` is not attribution, it is the column the access check runs
on: ``_ensure_archive_visible`` fails closed on NULL (404 for any caller
without ``archives:read_all``) and the list paths filter
``created_by_id == user.id``. So on a multi-user instance the tally reported
archives restored while their owner could neither list nor open them.
"""
def _entry(self, **overrides):
entry = {
"id": 77,
"filename": "benchy.3mf",
"file_size": 2048,
"content_hash": "abc123",
"started_at": "2026-03-01 10:00:00",
"created_at": "2026-03-01 10:00:00",
}
entry.update(overrides)
return entry
async def _user(self, db, username="alice"):
user = User(username=username, role="operator")
db.add(user)
await db.flush()
return user
@pytest.mark.asyncio
async def test_owner_is_carried_across(self, db_session):
user = await self._user(db_session)
tally = _CategoryTally()
await _service()._restore_archives(
db_session, {"archives": [self._entry(created_by_id=user.id)]}, False, tally, {}
)
await db_session.commit()
row = (await db_session.execute(select(PrintArchive))).scalar_one()
assert row.created_by_id == user.id
assert not any("owner cleared" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_an_unknown_owner_is_cleared_with_a_note_not_failed(self, db_session):
"""The archive is still worth having; an admin can reassign it."""
tally = _CategoryTally()
await _service()._restore_archives(
db_session, {"archives": [self._entry(created_by_id=4242)]}, False, tally, {}
)
await db_session.commit()
row = (await db_session.execute(select(PrintArchive))).scalar_one()
assert row.created_by_id is None
assert tally.restored == 1 and tally.failed == 0
assert any("owner cleared" in note and "archives:read_all" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_the_owner_note_is_emitted_once_for_many_rows(self, db_session):
tally = _CategoryTally()
archives = [
self._entry(id=1, content_hash="h1", filename="a.3mf", created_by_id=4242),
self._entry(id=2, content_hash="h2", filename="b.3mf", created_by_id=4243),
]
await _service()._restore_archives(db_session, {"archives": archives}, False, tally, {})
await db_session.commit()
assert sum(1 for note in _messages(tally) if "owner cleared" in note) == 1
@pytest.mark.asyncio
async def test_a_backup_without_the_key_still_restores(self, db_session):
"""Backups taken before the collector recorded it just can't know the owner."""
tally = _CategoryTally()
await _service()._restore_archives(db_session, {"archives": [self._entry()]}, False, tally, {})
await db_session.commit()
row = (await db_session.execute(select(PrintArchive))).scalar_one()
assert row.created_by_id is None
assert not any("owner cleared" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_overwrite_makes_the_local_owner_match_the_backup(self, db_session):
alice = await self._user(db_session, "alice")
bob = await self._user(db_session, "bob")
db_session.add(
PrintArchive(
filename="benchy.3mf",
file_path="/data/benchy.3mf",
file_size=2048,
content_hash="abc123",
started_at=datetime(2026, 3, 1, 10, 0, 0),
created_by_id=bob.id,
)
)
await db_session.commit()
await _service()._restore_archives(
db_session, {"archives": [self._entry(created_by_id=alice.id)]}, True, _CategoryTally(), {}
)
await db_session.commit()
row = (await db_session.execute(select(PrintArchive))).scalar_one()
assert row.created_by_id == alice.id
@pytest.mark.asyncio
async def test_overwrite_leaves_the_owner_alone_when_the_backup_predates_the_key(self, db_session):
"""A pre-#2656 commit must not blank the owner of a row that was fine.
The entry carries no ``created_by_id`` at all, so there is nothing to
write. Treating that as an explicit null inflicted the exact bug the
column was added to fix — a 404 for the owner — on rows the restore had
no business touching, silently, while still counting them restored.
"""
bob = await self._user(db_session, "bob")
db_session.add(
PrintArchive(
filename="benchy.3mf",
file_path="/data/benchy.3mf",
file_size=2048,
content_hash="abc123",
started_at=datetime(2026, 3, 1, 10, 0, 0),
created_by_id=bob.id,
)
)
await db_session.commit()
tally = _CategoryTally()
await _service()._restore_archives(db_session, {"archives": [self._entry()]}, True, tally, {})
await db_session.commit()
row = (await db_session.execute(select(PrintArchive))).scalar_one()
assert row.created_by_id == bob.id, "an old backup does not know the owner, so it must not clear one"
assert tally.restored == 1
@pytest.mark.asyncio
async def test_overwrite_leaves_deleted_at_alone_when_the_backup_predates_the_key(self, db_session):
"""The mirror case: an old commit must not un-delete, and must not claim to.
``archivesUndeleted`` reads the same absent value, so the un-delete was
not merely wrong but unannounced.
"""
db_session.add(
PrintArchive(
filename="benchy.3mf",
file_path="/data/benchy.3mf",
file_size=2048,
content_hash="abc123",
started_at=datetime(2026, 3, 1, 10, 0, 0),
deleted_at=datetime(2026, 3, 4, 8, 0, 0),
)
)
await db_session.commit()
tally = _CategoryTally()
await _service()._restore_archives(db_session, {"archives": [self._entry()]}, True, tally, {})
await db_session.commit()
row = (await db_session.execute(select(PrintArchive))).scalar_one()
assert row.deleted_at == datetime(2026, 3, 4, 8, 0, 0), "an old backup must not resurrect a deleted archive"
assert not any("visible again" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_overwrite_still_clears_an_owner_the_backup_explicitly_nulls(self, db_session):
"""Control: absent is ignored, but an explicit null is still honoured.
A current-format backup of an unowned archive has to be able to say so,
or overwrite stops meaning "make the local row match the backup".
"""
bob = await self._user(db_session, "bob")
db_session.add(
PrintArchive(
filename="benchy.3mf",
file_path="/data/benchy.3mf",
file_size=2048,
content_hash="abc123",
started_at=datetime(2026, 3, 1, 10, 0, 0),
created_by_id=bob.id,
)
)
await db_session.commit()
await _service()._restore_archives(
db_session, {"archives": [self._entry(created_by_id=None)]}, True, _CategoryTally(), {}
)
await db_session.commit()
row = (await db_session.execute(select(PrintArchive))).scalar_one()
assert row.created_by_id is None
@pytest.mark.asyncio
async def test_overwrite_still_undeletes_when_the_backup_explicitly_nulls(self, db_session):
"""Control for the deleted_at half, with the note that goes with it."""
db_session.add(
PrintArchive(
filename="benchy.3mf",
file_path="/data/benchy.3mf",
file_size=2048,
content_hash="abc123",
started_at=datetime(2026, 3, 1, 10, 0, 0),
deleted_at=datetime(2026, 3, 4, 8, 0, 0),
)
)
await db_session.commit()
tally = _CategoryTally()
await _service()._restore_archives(db_session, {"archives": [self._entry(deleted_at=None)]}, True, tally, {})
await db_session.commit()
row = (await db_session.execute(select(PrintArchive))).scalar_one()
assert row.deleted_at is None
assert any("visible again" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_owner_survives_collect_then_restore(self, db_session):
"""Both halves, because each looks harmless alone.
The collector never wrote the key, so there was nothing for the restore
to carry across even once it wanted to.
"""
from backend.app.services.github_backup import github_backup_service
user = await self._user(db_session)
db_session.add(
PrintArchive(
filename="owned.3mf",
file_path="",
file_size=1024,
content_hash="hash-owned",
started_at=datetime(2026, 3, 1, 10, 0, 0),
created_by_id=user.id,
)
)
await db_session.commit()
files: dict = {}
await github_backup_service._collect_archives(db_session, files)
payload = files[ARCHIVES_PATH]
assert payload["archives"][0]["created_by_id"] == user.id
await db_session.execute(PrintArchive.__table__.delete())
await db_session.commit()
await _service()._restore_archives(db_session, payload, False, _CategoryTally(), {})
await db_session.commit()
row = (await db_session.execute(select(PrintArchive))).scalar_one()
assert row.created_by_id == user.id, "a restored archive its owner cannot see is not restored"
class TestCategoryPathMapping:
def setup_method(self):
self.service = _service()
self.available = [
"backup_metadata.json",
SETTINGS_PATH,
SPOOLS_PATH,
SPOOL_USAGE_PATH,
ARCHIVES_PATH,
"kprofiles/SERIAL1/0.4.json",
"kprofiles/SERIAL1/0.8.json",
"cloud_profiles/filament.json",
]
def test_spools_includes_usage_history(self):
paths = self.service._category_paths(RestoreCategory.SPOOLS, self.available)
assert paths == [SPOOLS_PATH, SPOOL_USAGE_PATH]
def test_kprofiles_globs_all_serials_and_nozzles(self):
paths = self.service._category_paths(RestoreCategory.KPROFILES, self.available)
assert paths == ["kprofiles/SERIAL1/0.4.json", "kprofiles/SERIAL1/0.8.json"]
def test_absent_paths_are_omitted(self):
paths = self.service._category_paths(RestoreCategory.SETTINGS, ["backup_metadata.json"])
assert paths == []
def test_cloud_profiles_are_not_a_restore_category(self):
assert "cloud_profiles" not in {c.value for c in RestoreCategory}
class TestMutex:
@pytest.mark.asyncio
async def test_restore_refuses_while_a_backup_is_running(self):
service = _service()
with patch("backend.app.services.github_backup.github_backup_service") as backup:
backup.is_running = True
result = await service.run_restore(1, "HEAD", [RestoreCategory.SPOOLS])
assert result["success"] is False
assert "backup is currently running" in result["message"]
@pytest.mark.asyncio
async def test_restore_refuses_while_another_restore_is_running(self):
service = _service()
service._running_restore = True
result = await service.run_restore(1, "HEAD", [RestoreCategory.SPOOLS])
assert result["success"] is False
assert "restore is already running" in result["message"]
@pytest.mark.asyncio
async def test_backup_refuses_while_a_restore_is_running(self):
from backend.app.services.github_backup import GitHubBackupService
backup_service = GitHubBackupService()
with patch("backend.app.services.github_restore.github_restore_service") as restore:
restore.is_running = True
result = await backup_service.run_backup(1, trigger="manual")
assert result["success"] is False
assert "restore is currently running" in result["message"]
class TestMqttRelayReconfigure:
"""Restoring mqtt_* rows has to reach the live relay, not just the table."""
@pytest.mark.asyncio
async def test_reconfigures_from_the_committed_rows(self, db_session):
db_session.add(Settings(key="mqtt_enabled", value="true"))
db_session.add(Settings(key="mqtt_broker", value="restored.local"))
db_session.add(Settings(key="mqtt_port", value="8883"))
db_session.add(Settings(key="mqtt_use_tls", value="true"))
# Never restorable (credential blocklist), so it comes from the row that
# was already there.
db_session.add(Settings(key="mqtt_password", value="kept"))
await db_session.commit()
tally = _CategoryTally()
relay = MagicMock()
relay.configure = AsyncMock(return_value=True)
with patch("backend.app.services.mqtt_relay.mqtt_relay", relay):
await _service()._reconfigure_mqtt_relay(db_session, {"mqtt_broker"}, tally)
relay.configure.assert_awaited_once()
sent = relay.configure.await_args.args[0]
assert sent["mqtt_enabled"] is True
assert sent["mqtt_broker"] == "restored.local"
assert sent["mqtt_port"] == 8883
assert sent["mqtt_use_tls"] is True
assert sent["mqtt_password"] == "kept"
assert sent["mqtt_topic_prefix"] == "bambuddy"
assert tally.notes == []
@pytest.mark.asyncio
async def test_no_reconnect_when_no_mqtt_key_was_written(self, db_session):
"""configure() tears the connection down, so don't call it for a theme change."""
tally = _CategoryTally()
relay = MagicMock()
relay.configure = AsyncMock()
with patch("backend.app.services.mqtt_relay.mqtt_relay", relay):
await _service()._reconfigure_mqtt_relay(db_session, {"currency", "theme"}, tally)
relay.configure.assert_not_awaited()
@pytest.mark.asyncio
async def test_broker_failure_is_noted_not_fatal(self, db_session):
tally = _CategoryTally()
relay = MagicMock()
relay.configure = AsyncMock(side_effect=OSError("no route to broker"))
with patch("backend.app.services.mqtt_relay.mqtt_relay", relay):
await _service()._reconfigure_mqtt_relay(db_session, {"mqtt_enabled"}, tally)
assert any("restart Bambuddy" in note for note in _messages(tally))
@pytest.mark.asyncio
async def test_restore_settings_reports_the_keys_it_wrote(self, db_session):
db_session.add(Settings(key="mqtt_broker", value="old.local"))
await db_session.commit()
written: set[str] = set()
payload = {
"settings": {
"mqtt_broker": "new.local",
"currency": "EUR",
"mqtt_password": "leaked",
"auth_enabled": "false",
}
}
await _service()._restore_settings(
db_session, payload, overwrite=True, tally=_CategoryTally(), keys_written=written
)
# Skipped keys are not "written", or a blocked mqtt_password would
# trigger a pointless reconnect.
assert written == {"mqtt_broker", "currency"}
@pytest.mark.asyncio
async def test_keys_skipped_for_overwrite_off_are_not_reported(self, db_session):
db_session.add(Settings(key="mqtt_broker", value="old.local"))
await db_session.commit()
written: set[str] = set()
await _service()._restore_settings(
db_session,
{"settings": {"mqtt_broker": "new.local"}},
overwrite=False,
tally=_CategoryTally(),
keys_written=written,
)
assert written == set()
@pytest.mark.asyncio
async def test_a_refused_mqtt_enabled_is_not_reported_as_written(self, db_session):
"""So the relay reconfigures from the *local* mqtt_enabled, not the backup's.
The companion rule refuses ``mqtt_enabled`` when the backup's password
cannot come across and there is none stored locally. It must not then
appear in ``keys_written``, or _reconfigure_mqtt_relay would be asked to
bring up a broker connection the restore deliberately declined to enable.
"""
written: set[str] = set()
await _service()._restore_settings(
db_session,
{"settings": {"mqtt_enabled": "true", "mqtt_password": "refused", "mqtt_broker": "new.local"}},
overwrite=True,
tally=_CategoryTally(),
keys_written=written,
)
assert written == {"mqtt_broker"}
class TestApplyOrdering:
"""_apply must not hold SQLite's write transaction across the MQTT phase."""
def _recording_service(self, calls: list[str]):
service = _service()
# Sync side effects on purpose: an AsyncMock returns a coroutine its
# side_effect hands back rather than awaiting it, so an async recorder
# would never run.
service._restore_archives = AsyncMock(side_effect=lambda *a, **k: calls.append("archives"))
service._restore_spools = AsyncMock(side_effect=lambda *a, **k: calls.append("spools"))
service._restore_settings = AsyncMock(side_effect=lambda *a, **k: calls.append("settings"))
service._restore_kprofiles = AsyncMock(side_effect=lambda *a, **k: calls.append("kprofiles"))
return service
@pytest.mark.asyncio
async def test_commits_database_categories_before_talking_to_printers(self):
"""get_kprofiles is 3 x 5 s per printer/nozzle; busy_timeout is 15 s."""
calls: list[str] = []
service = self._recording_service(calls)
db = MagicMock()
db.commit = AsyncMock(side_effect=lambda: calls.append("commit"))
await service._apply(
db,
{},
[RestoreCategory.ARCHIVES, RestoreCategory.SPOOLS, RestoreCategory.KPROFILES],
False,
)
assert calls == ["archives", "spools", "commit", "kprofiles"]
@pytest.mark.asyncio
async def test_does_not_split_the_transaction_without_kprofiles(self):
"""A database-only restore stays one transaction, committed by run_restore."""
calls: list[str] = []
service = self._recording_service(calls)
db = MagicMock()
db.commit = AsyncMock(side_effect=lambda: calls.append("commit"))
await service._apply(db, {}, [RestoreCategory.ARCHIVES, RestoreCategory.SETTINGS], False)
assert calls == ["archives", "settings"]
assert db.commit.await_count == 0
class TestKprofilePhaseFailure:
"""The K-profile phase runs after _apply has committed everything else.
So an exception there used to reach run_restore's handler, which reports
``success: False`` with an empty ``results`` — over archive, spool and
settings rows that are durable on disk. The honest-reporting theme of this
feature inverted on exactly the path where it matters, and the post-commit
MQTT reconfigure (downstream of the raise, inside the same try) was skipped,
leaving the relay pointed at the pre-restore broker.
"""
_SETTINGS = {"version": "1.0", "settings": {"mqtt_broker": "restored.local", "currency": "EUR"}}
def _payload(self, profiles=None):
return {
SETTINGS_PATH: dict(self._SETTINGS),
"kprofiles/00M09A123456789/0.4.json": {
"profiles": [{"filament_id": "GFA00", "name": "Bambu PLA"}] if profiles is None else profiles
},
}
def _session_patch(self, db_session):
cm = AsyncMock()
cm.__aenter__ = AsyncMock(return_value=db_session)
cm.__aexit__ = AsyncMock(return_value=None)
return patch("backend.app.services.github_restore.async_session", return_value=cm)
async def _configured_service(self, db_session, payload):
from backend.app.models.github_backup import GitHubBackupConfig
config = GitHubBackupConfig(repository_url="https://github.com/o/r", access_token="tok", provider="github")
db_session.add(config)
await db_session.commit()
service = _service()
service._resolve_ref = AsyncMock(return_value=("a" * 40, "", None))
service._read_categories = AsyncMock(return_value=(payload, ""))
return service, config.id
@pytest.mark.asyncio
async def test_the_committed_categories_are_still_reported(self, db_session):
service = _service()
service._restore_kprofiles = AsyncMock(side_effect=RuntimeError("mqtt exploded"))
results = await service._apply(
db_session,
self._payload(),
[RestoreCategory.SETTINGS, RestoreCategory.KPROFILES],
False,
)
assert results[RestoreCategory.SETTINGS.value].restored == 2
rows = {s.key: s.value for s in (await db_session.execute(select(Settings))).scalars().all()}
assert rows == {"mqtt_broker": "restored.local", "currency": "EUR"}, "committed before the phase that failed"
@pytest.mark.asyncio
async def test_the_failure_is_counted_and_explained(self, db_session):
service = _service()
service._restore_kprofiles = AsyncMock(side_effect=RuntimeError("mqtt exploded"))
results = await service._apply(
db_session,
self._payload(profiles=[{"filament_id": "GFA00"}, {"filament_id": "GFB99"}]),
[RestoreCategory.SETTINGS, RestoreCategory.KPROFILES],
False,
)
tally = results[RestoreCategory.KPROFILES.value]
assert tally.failed == 2, "every profile the payload carried is unaccounted for"
assert tally.restored == 0
assert _codes(tally) == ["kprofilesStepFailed"]
assert tally.notes[0]["params"]["reason"] == "mqtt exploded"
@pytest.mark.asyncio
async def test_the_relay_is_reconfigured_even_though_the_phase_failed(self, db_session):
"""The reconfigure sits downstream of the raise in run_restore's try."""
service, config_id = await self._configured_service(db_session, self._payload())
service._restore_kprofiles = AsyncMock(side_effect=RuntimeError("mqtt exploded"))
relay = MagicMock()
relay.configure = AsyncMock(return_value=True)
with self._session_patch(db_session), patch("backend.app.services.mqtt_relay.mqtt_relay", relay):
result = await service.run_restore(
config_id, "a" * 40, [RestoreCategory.SETTINGS, RestoreCategory.KPROFILES]
)
assert result["success"] is True
assert result["results"][RestoreCategory.SETTINGS.value]["restored"] == 2
assert result["results"][RestoreCategory.KPROFILES.value]["failed"] == 1
relay.configure.assert_awaited_once()
assert relay.configure.await_args.args[0]["mqtt_broker"] == "restored.local"
@pytest.mark.asyncio
async def test_a_failure_before_the_commit_still_reports_nothing_restored(self, db_session):
"""Control: rolling back and saying so is right when nothing landed."""
service, config_id = await self._configured_service(db_session, self._payload())
service._restore_settings = AsyncMock(side_effect=RuntimeError("read failed"))
relay = MagicMock()
relay.configure = AsyncMock(return_value=True)
with self._session_patch(db_session), patch("backend.app.services.mqtt_relay.mqtt_relay", relay):
result = await service.run_restore(
config_id, "a" * 40, [RestoreCategory.SETTINGS, RestoreCategory.KPROFILES]
)
assert result["success"] is False
assert result["results"] == {}
assert (await db_session.execute(select(Settings))).scalars().first() is None
relay.configure.assert_not_awaited()
@pytest.mark.asyncio
async def test_a_malformed_profiles_value_is_a_skipped_category_not_a_raise(self, db_session, printer_factory):
"""Belt-and-braces: the pre-loop count ran ahead of the per-call guards.
``sum(len(c.get("profiles") or []) ...)`` raises TypeError on a
hand-edited or truncated backup whose ``profiles`` is not a list — and it
raises after the database categories are already on disk.
"""
await printer_factory(serial_number="00M09A123456789")
client = MagicMock()
client.state.connected = True
client.set_kprofiles_batch = MagicMock(return_value="7")
client.get_kprofiles = AsyncMock(return_value=[])
tally = _CategoryTally()
with patch("backend.app.services.github_restore.printer_manager") as manager:
manager.get_client = MagicMock(return_value=client)
await _service()._restore_kprofiles(db_session, self._payload(profiles=5), tally)
client.set_kprofiles_batch.assert_not_called()
assert (tally.restored, tally.failed) == (0, 0)
assert "kprofilesStepFailed" not in _codes(tally)
class TestResolveRef:
@pytest.mark.asyncio
async def test_concrete_sha_passes_through_without_an_api_call(self):
service = _service()
service.list_commits = AsyncMock()
config = MagicMock(branch="main")
resolved, error, commit = await service._resolve_ref(config, "abc1234")
assert resolved == "abc1234"
assert error == ""
# Nothing was fetched, so there is no entry to describe it with.
assert commit is None
service.list_commits.assert_not_awaited()
@pytest.mark.asyncio
async def test_head_resolves_to_the_tip_sha(self):
service = _service()
service.list_commits = AsyncMock(
return_value={"success": True, "commits": [{"sha": "tipsha1"}, {"sha": "older"}]}
)
config = MagicMock(branch="main")
resolved, error, commit = await service._resolve_ref(config, "HEAD")
assert resolved == "tipsha1"
assert error == ""
# Handed back so preview does not list commits a second time just to
# describe the one it already fetched.
assert commit == {"sha": "tipsha1"}
@pytest.mark.asyncio
async def test_empty_history_is_an_error(self):
service = _service()
service.list_commits = AsyncMock(return_value={"success": True, "commits": []})
config = MagicMock(branch="main")
resolved, error, commit = await service._resolve_ref(config, "HEAD")
assert resolved is None
assert "no commits" in error
assert commit is None
class TestDescribeCommit:
"""A preview that says `commit: null` gives the user no idea what they picked."""
def _config(self):
return MagicMock(branch="main", provider="github", repository_url="https://github.com/o/r", access_token="t")
def _entry(self, sha: str):
return {"sha": sha, "message": "Bambuddy backup", "author": "Bambuddy", "date": "2026-07-01T10:00:00Z"}
@pytest.mark.asyncio
async def test_an_abbreviated_ref_matches_a_full_sha_in_the_window(self):
"""REF_PATTERN accepts 7 characters; providers return 40.
The old exact `==` therefore never matched an abbreviated ref, even when
the commit was right there in the top 20.
"""
service = _service()
full = "abc1234" + "0" * 33
service.list_commits = AsyncMock(return_value={"success": True, "commits": [self._entry(full)]})
found = await service._describe_commit(self._config(), "abc1234")
assert found is not None
assert found["sha"] == full
@pytest.mark.asyncio
async def test_a_full_sha_matches_an_abbreviated_entry(self):
service = _service()
service.list_commits = AsyncMock(return_value={"success": True, "commits": [self._entry("abc1234")]})
found = await service._describe_commit(self._config(), "abc1234" + "0" * 33)
assert found is not None
@pytest.mark.asyncio
async def test_a_commit_outside_the_window_is_fetched_directly(self):
service = _service()
service.list_commits = AsyncMock(return_value={"success": True, "commits": [self._entry("f" * 40)]})
backend = MagicMock()
backend.get_commit = AsyncMock(return_value={"success": True, "commit": self._entry("old" + "0" * 37)})
with patch("backend.app.services.github_restore.get_provider_backend", return_value=backend):
found = await service._describe_commit(self._config(), "old" + "0" * 37)
assert found["sha"] == "old" + "0" * 37
backend.get_commit.assert_awaited_once()
@pytest.mark.asyncio
async def test_a_direct_lookup_failure_is_not_fatal(self):
"""It is a subject line: render the preview without it."""
service = _service()
service.list_commits = AsyncMock(return_value={"success": True, "commits": []})
backend = MagicMock()
backend.get_commit = AsyncMock(return_value={"success": False, "message": "boom", "commit": None})
with patch("backend.app.services.github_restore.get_provider_backend", return_value=backend):
assert await service._describe_commit(self._config(), "a" * 40) is None
@pytest.mark.asyncio
async def test_the_window_scan_is_not_run_twice(self):
"""_resolve_ref already listed commits for HEAD; preview reuses that."""
service = _service()
tip = self._entry("t" * 40)
service.list_commits = AsyncMock(return_value={"success": True, "commits": [tip]})
resolved, _, commit = await service._resolve_ref(self._config(), "HEAD")
assert resolved == "t" * 40
assert commit == tip
assert service.list_commits.await_count == 1
Reference in New Issue View Git Blame Copy Permalink