|
|
@@ -0,0 +1,257 @@
|
|
|
+"""Which prints are worth an FTPS sweep, and which are not (#2780).
|
|
|
+
|
|
|
+The gate this module guards is one-sided on purpose, and both sides matter:
|
|
|
+
|
|
|
+* Missing it costs ~110 doomed FTP connections per print and an archive card
|
|
|
+ that is blank with no stated reason -- the reported bug.
|
|
|
+* Over-applying it costs archives that work today. A printer that never
|
|
|
+ publishes ``sdcard`` and never had a ``project_file`` reach us must sweep
|
|
|
+ exactly as before, or the fix is a regression for everyone else.
|
|
|
+
|
|
|
+So the tests below spend most of their weight on the second failure mode.
|
|
|
+"""
|
|
|
+
|
|
|
+import pytest
|
|
|
+
|
|
|
+from backend.app.services.print_storage import (
|
|
|
+ REASON_INTERNAL_STORAGE,
|
|
|
+ REASON_NO_EXTERNAL_STORAGE,
|
|
|
+ external_storage_present,
|
|
|
+ last_print_storage_verdict,
|
|
|
+ print_file_reachable_over_ftp,
|
|
|
+ url_is_external_storage,
|
|
|
+)
|
|
|
+
|
|
|
+pytestmark = pytest.mark.unit
|
|
|
+
|
|
|
+
|
|
|
+class FakeState:
|
|
|
+ """Stand-in for PrinterState with only the fields the helper reads."""
|
|
|
+
|
|
|
+ def __init__(self, current_project_url=None, sdcard=False, sdcard_reported=False, last_project_url=None):
|
|
|
+ self.current_project_url = current_project_url
|
|
|
+ # Defaults to the per-print value: for every test that does not care
|
|
|
+ # about the distinction, the two readings agree.
|
|
|
+ self.last_project_url = current_project_url if last_project_url is None else last_project_url
|
|
|
+ self.sdcard = sdcard
|
|
|
+ self.sdcard_reported = sdcard_reported
|
|
|
+
|
|
|
+
|
|
|
+class TestUrlScheme:
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "url",
|
|
|
+ [
|
|
|
+ "ftp://Benchy.gcode.3mf",
|
|
|
+ # Real dispatches carry names with spaces and non-ASCII; the scheme
|
|
|
+ # is all that is being read and none of that should disturb it.
|
|
|
+ "ftp://Halterung Kühlschrank V2.gcode.3mf",
|
|
|
+ "FTP://Benchy.gcode.3mf",
|
|
|
+ ],
|
|
|
+ )
|
|
|
+ def test_ftp_means_external_storage(self, url):
|
|
|
+ assert url_is_external_storage(url) is True
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "url",
|
|
|
+ [
|
|
|
+ # The scheme every H2C and P2S dispatch in #2780's bundle carried,
|
|
|
+ # 35 out of 35.
|
|
|
+ "brtc://emmc/169356_204314.STEP.gcode.3mf",
|
|
|
+ "brtc://emmc/Benchy.gcode.3mf",
|
|
|
+ ],
|
|
|
+ )
|
|
|
+ def test_brtc_means_internal_storage(self, url):
|
|
|
+ assert url_is_external_storage(url) is False
|
|
|
+
|
|
|
+ def test_an_unknown_scheme_is_not_assumed_reachable(self):
|
|
|
+ """Matching the reachable value, not the unreachable one.
|
|
|
+
|
|
|
+ If Bambu ships a third scheme, the safe reading is "somewhere we can't
|
|
|
+ see", not "fine" -- an unrecognised scheme that read as reachable would
|
|
|
+ put the storm straight back.
|
|
|
+ """
|
|
|
+ assert url_is_external_storage("sftp://Benchy.3mf") is False
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("url", [None, "", "Benchy.gcode.3mf"])
|
|
|
+ def test_no_usable_url_declines_to_answer(self, url):
|
|
|
+ """None is a third answer and must not collapse into False."""
|
|
|
+ assert url_is_external_storage(url) is None
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("url", [12345, [], {}, object()])
|
|
|
+ def test_a_non_string_url_declines_too(self, url):
|
|
|
+ """The value arrives straight off the wire, so it is whatever the
|
|
|
+ sender put there. Truth-testing alone would let a non-string fall
|
|
|
+ through to the scheme comparison and read as internal storage --
|
|
|
+ which is a silent skip of a sweep that should have run.
|
|
|
+ """
|
|
|
+ assert url_is_external_storage(url) is None
|
|
|
+
|
|
|
+
|
|
|
+class TestSweepIsSkipped:
|
|
|
+ def test_a_print_kept_on_internal_storage(self):
|
|
|
+ verdict = print_file_reachable_over_ftp(FakeState(current_project_url="brtc://emmc/Benchy.gcode.3mf"))
|
|
|
+
|
|
|
+ assert verdict.reachable is False
|
|
|
+ assert verdict.reason == REASON_INTERNAL_STORAGE
|
|
|
+
|
|
|
+ def test_a_printer_that_says_its_slot_is_empty(self):
|
|
|
+ """#2780's H2C: `sdcard` False for three weeks, 800 clean FTPS
|
|
|
+ connections, and a 550 on every single path it asked for."""
|
|
|
+ verdict = print_file_reachable_over_ftp(FakeState(sdcard=False, sdcard_reported=True))
|
|
|
+
|
|
|
+ assert verdict.reachable is False
|
|
|
+ assert verdict.reason == REASON_NO_EXTERNAL_STORAGE
|
|
|
+
|
|
|
+
|
|
|
+class TestSweepStillRuns:
|
|
|
+ """The regression guard. Every case here worked before the gate existed."""
|
|
|
+
|
|
|
+ def test_a_print_on_external_storage(self):
|
|
|
+ assert print_file_reachable_over_ftp(FakeState(current_project_url="ftp://Benchy.gcode.3mf")).reachable
|
|
|
+
|
|
|
+ def test_a_printer_that_never_mentioned_its_card(self):
|
|
|
+ """Silence is not evidence.
|
|
|
+
|
|
|
+ `sdcard` defaults to False, so a printer whose firmware simply never
|
|
|
+ publishes the field looks identical to an empty slot unless the
|
|
|
+ "did it ever say so" flag is honoured. Reading the default as an
|
|
|
+ answer would skip the sweep for every one of them.
|
|
|
+ """
|
|
|
+ assert print_file_reachable_over_ftp(FakeState(sdcard=False, sdcard_reported=False)).reachable
|
|
|
+
|
|
|
+ def test_a_printer_with_a_card_and_no_dispatch_seen(self):
|
|
|
+ """Some brokers refuse the request-topic subscription, so no URL ever
|
|
|
+ arrives. That install must behave exactly as it did before."""
|
|
|
+ assert print_file_reachable_over_ftp(FakeState(sdcard=True, sdcard_reported=True)).reachable
|
|
|
+
|
|
|
+ def test_an_explicit_ftp_url_outranks_a_disagreeing_card_flag(self):
|
|
|
+ """A false skip is a regression; a needless sweep is only slow.
|
|
|
+
|
|
|
+ When the dispatcher says the file went to external storage, believe
|
|
|
+ the specific claim over the general one.
|
|
|
+ """
|
|
|
+ state = FakeState(current_project_url="ftp://Benchy.gcode.3mf", sdcard=False, sdcard_reported=True)
|
|
|
+
|
|
|
+ assert print_file_reachable_over_ftp(state).reachable
|
|
|
+
|
|
|
+ def test_no_state_at_all(self):
|
|
|
+ """Printer not connected, or status not yet populated."""
|
|
|
+ assert print_file_reachable_over_ftp(None).reachable
|
|
|
+
|
|
|
+ def test_a_state_missing_the_fields_entirely(self):
|
|
|
+ """The helper is duck-typed, and a PrinterState from a pickled or
|
|
|
+ partially-constructed source may predate these fields."""
|
|
|
+
|
|
|
+ class Bare:
|
|
|
+ pass
|
|
|
+
|
|
|
+ assert print_file_reachable_over_ftp(Bare()).reachable
|
|
|
+
|
|
|
+
|
|
|
+class TestReasonIsAlwaysPresentWhenUnreachable:
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "state",
|
|
|
+ [
|
|
|
+ FakeState(current_project_url="brtc://emmc/x.3mf"),
|
|
|
+ FakeState(sdcard=False, sdcard_reported=True),
|
|
|
+ ],
|
|
|
+ )
|
|
|
+ def test_unreachable_carries_a_reason(self, state):
|
|
|
+ """The reason crosses into the API and picks the banner text. An
|
|
|
+ unreachable verdict without one would render the generic advice --
|
|
|
+ which is the wrong advice, and the whole point of the change."""
|
|
|
+ verdict = print_file_reachable_over_ftp(state)
|
|
|
+
|
|
|
+ assert verdict.reachable is False
|
|
|
+ assert verdict.reason
|
|
|
+
|
|
|
+ def test_reachable_carries_no_reason(self):
|
|
|
+ assert print_file_reachable_over_ftp(FakeState(sdcard=True, sdcard_reported=True)).reason is None
|
|
|
+
|
|
|
+
|
|
|
+class TestTheGateUsesThePerPrintUrlOnly:
|
|
|
+ """A stale URL must never gate a sweep.
|
|
|
+
|
|
|
+ ``current_project_url`` is cleared when a print ends; ``last_project_url``
|
|
|
+ is sticky for reporting. The gate reads only the first, and that is
|
|
|
+ load-bearing rather than tidiness: 18% of the print starts in #2780's
|
|
|
+ support bundle (14 of 79) had no ``project_file`` on the request topic at
|
|
|
+ all -- touchscreen reprints, restart recovery, anything Bambuddy did not
|
|
|
+ see dispatched. If those inherited the previous job's destination, a
|
|
|
+ printer that ran one Studio print to internal storage would skip the FTPS
|
|
|
+ sweep for every subsequent screen-started print, losing archives that work
|
|
|
+ today.
|
|
|
+
|
|
|
+ The asymmetry is what makes it worth pinning: a stale ``ftp://`` costs
|
|
|
+ only a pointless sweep, while a stale ``brtc://`` costs an archive.
|
|
|
+ """
|
|
|
+
|
|
|
+ def test_a_print_with_no_dispatch_of_its_own_still_sweeps(self):
|
|
|
+ """The previous print went to internal storage; this one Bambuddy
|
|
|
+ never saw dispatched. Unknown, so sweep."""
|
|
|
+ state = FakeState(
|
|
|
+ current_project_url=None,
|
|
|
+ last_project_url="brtc://emmc/previous.gcode.3mf",
|
|
|
+ sdcard=True,
|
|
|
+ sdcard_reported=True,
|
|
|
+ )
|
|
|
+
|
|
|
+ assert print_file_reachable_over_ftp(state).reachable
|
|
|
+
|
|
|
+ def test_the_sticky_reading_still_reports_it(self):
|
|
|
+ """The diagnostic is normally run after the print that prompted it, so
|
|
|
+ it needs the answer the gate has rightly forgotten."""
|
|
|
+ state = FakeState(
|
|
|
+ current_project_url=None,
|
|
|
+ last_project_url="brtc://emmc/previous.gcode.3mf",
|
|
|
+ sdcard=True,
|
|
|
+ sdcard_reported=True,
|
|
|
+ )
|
|
|
+
|
|
|
+ verdict = last_print_storage_verdict(state)
|
|
|
+
|
|
|
+ assert verdict.reachable is False
|
|
|
+ assert verdict.reason == REASON_INTERNAL_STORAGE
|
|
|
+
|
|
|
+ def test_an_empty_slot_is_reported_by_both(self):
|
|
|
+ """Not URL-derived, so clearing the per-print value changes nothing."""
|
|
|
+ state = FakeState(sdcard=False, sdcard_reported=True)
|
|
|
+
|
|
|
+ assert print_file_reachable_over_ftp(state).reason == REASON_NO_EXTERNAL_STORAGE
|
|
|
+ assert last_print_storage_verdict(state).reason == REASON_NO_EXTERNAL_STORAGE
|
|
|
+
|
|
|
+ def test_the_sticky_reading_never_gates_a_sweep(self):
|
|
|
+ """Guard against someone swapping the two back: if the gate ever reads
|
|
|
+ the sticky field, the case above starts failing -- and so does this."""
|
|
|
+ import inspect
|
|
|
+
|
|
|
+ from backend.app.services import print_storage
|
|
|
+
|
|
|
+ source = inspect.getsource(print_storage.print_file_reachable_over_ftp)
|
|
|
+ assert "current_project_url" in source
|
|
|
+ assert "last_project_url" not in source.split('"""')[-1]
|
|
|
+
|
|
|
+
|
|
|
+class TestTimelapseUsesTheNarrowerRule:
|
|
|
+ """The printer writes its timelapse to the card itself.
|
|
|
+
|
|
|
+ Where the *sliced file* went says nothing about whether a video exists, so
|
|
|
+ gating the timelapse scan on the URL would silently stop finding videos on
|
|
|
+ every H2C and P2S that has a card in -- a new bug, introduced by the fix
|
|
|
+ for this one.
|
|
|
+ """
|
|
|
+
|
|
|
+ def test_internal_storage_does_not_suppress_the_timelapse_scan(self):
|
|
|
+ state = FakeState(current_project_url="brtc://emmc/x.3mf", sdcard=True, sdcard_reported=True)
|
|
|
+
|
|
|
+ assert print_file_reachable_over_ftp(state).reachable is False
|
|
|
+ assert external_storage_present(state) is True
|
|
|
+
|
|
|
+ def test_an_empty_slot_does_suppress_it(self):
|
|
|
+ assert external_storage_present(FakeState(sdcard=False, sdcard_reported=True)) is False
|
|
|
+
|
|
|
+ def test_silence_does_not(self):
|
|
|
+ assert external_storage_present(FakeState(sdcard=False, sdcard_reported=False)) is True
|
|
|
+
|
|
|
+ def test_no_state_does_not(self):
|
|
|
+ assert external_storage_present(None) is True
|