mirror of
https://github.com/Matysh/houseplan-card
synced 2026-07-31 16:38:31 +00:00
Matysh
3e976562ff
Owner's spec (2026-07-29), agreed points: one 'Run' action covering the three runnable domains of HA (a script is the idiomatic 'action' — with automations alone people would build trigger-less dummies); the confirm checkbox guards BOTH toggle and run; covers and valves join the card-wide toggle so curtains work natively. - marker.tap_action gains 'run'; marker.tap_target (schema-bounded to automation./script./scene. ids); marker.tap_confirm. - the dialog: a searchable picker over the three domains (friendly name + kind), save refuses a run action without a target, a vanished target gets a warning hint; the checkbox shows for any actionable tap (explicit or effective-default toggle). - the tap: automation.trigger / script.turn_on / scene.turn_on, started/ error toasts; with confirm on — our own dialog (not window.confirm, it must work on a wall tablet), Esc/backdrop/Cancel = no call. The guard covers the controls-toggle path too. - 'run' is explicit-only by construction: it needs a per-marker target, so it can never arrive as a card-wide default. - covers: the old test pinned 'garage stays shut' — that intent survives as COVER_GUARDED_CLASSES (garage/door/gate stay out of the CARD-WIDE toggle; an explicit per-device toggle remains the owner's conscious choice). Locks/alarms stay forbidden everywhere, run included is not affected — we do not inspect automation contents, same trust as HA's own Run button. Tests: unit resolveTapAction/runServiceFor + cover guard, backend schema parity picks 'run' automatically + tap_target bounds, smoke_tap_run with 11 assertions (picker, search, save guard, confirm cancel/ok, per-domain services, missing target). smoke_tap_ctx: 4 options now. Inventory: 148 / 52 / 43 / 72.
926 lines
36 KiB
Python
926 lines
36 KiB
Python
"""Unit tests for the pure House Plan validation (validation.py is loaded by path,
|
|
without importing the HA integration package)."""
|
|
import importlib.util
|
|
import os
|
|
|
|
import pytest
|
|
import voluptuous as vol
|
|
|
|
_PATH = os.path.join(
|
|
os.path.dirname(os.path.dirname(__file__)),
|
|
"custom_components", "houseplan", "validation.py",
|
|
)
|
|
_spec = importlib.util.spec_from_file_location("hp_validation", _PATH)
|
|
v = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(v)
|
|
|
|
|
|
def _load_pure(name):
|
|
"""Load one pure module of the integration without importing the package.
|
|
|
|
custom_components/houseplan/__init__.py pulls in Home Assistant, which the
|
|
local sandbox does not have; but plans.py is deliberately pure, so a tiny
|
|
stub package lets it keep its normal relative imports.
|
|
"""
|
|
import sys
|
|
import types
|
|
|
|
pkg_dir = os.path.join(
|
|
os.path.dirname(os.path.dirname(__file__)), "custom_components", "houseplan"
|
|
)
|
|
pkg = sys.modules.get("hp_pure")
|
|
if pkg is None:
|
|
pkg = types.ModuleType("hp_pure")
|
|
pkg.__path__ = [pkg_dir]
|
|
sys.modules["hp_pure"] = pkg
|
|
for dep in ("const", "validation"):
|
|
sp = importlib.util.spec_from_file_location(
|
|
f"hp_pure.{dep}", os.path.join(pkg_dir, f"{dep}.py")
|
|
)
|
|
mod = importlib.util.module_from_spec(sp)
|
|
sys.modules[f"hp_pure.{dep}"] = mod
|
|
sp.loader.exec_module(mod)
|
|
sp = importlib.util.spec_from_file_location(
|
|
f"hp_pure.{name}", os.path.join(pkg_dir, f"{name}.py")
|
|
)
|
|
mod = importlib.util.module_from_spec(sp)
|
|
sys.modules[f"hp_pure.{name}"] = mod
|
|
sp.loader.exec_module(mod)
|
|
return mod
|
|
|
|
|
|
plans = _load_pure("plans")
|
|
const = importlib.import_module("hp_pure.const")
|
|
|
|
|
|
def test_sanitize_marker_id():
|
|
assert v.sanitize_marker_id("../etc/passwd") == "_etc_passwd"
|
|
assert v.sanitize_marker_id("..") == "misc" # pure traversal → misc
|
|
assert v.sanitize_marker_id(".") == "misc"
|
|
assert v.sanitize_marker_id("") == "misc"
|
|
assert len(v.sanitize_marker_id("a" * 200)) == 64
|
|
|
|
|
|
def test_sanitize_filename_strips_path():
|
|
assert v.sanitize_filename("/a/b/c/manual.pdf") == "manual.pdf"
|
|
assert v.sanitize_filename("..\\..\\evil.pdf") == "evil.pdf" # backslashes = a path
|
|
assert v.sanitize_filename("...hidden.pdf") == "hidden.pdf" # leading dots stripped
|
|
|
|
|
|
def test_file_ext():
|
|
assert v.file_ext("manual.PDF") == "pdf"
|
|
assert v.file_ext("a/b/x.png") == "png"
|
|
assert v.file_ext("noext") == ""
|
|
|
|
|
|
def test_valid_space_id():
|
|
assert v.valid_space_id("f1")
|
|
assert v.valid_space_id("floor-2_a")
|
|
assert not v.valid_space_id("Floor 1")
|
|
assert not v.valid_space_id("a" * 65)
|
|
assert not v.valid_space_id("../x")
|
|
|
|
|
|
def test_room_schema_poly_or_rect():
|
|
v.ROOM_SCHEMA({"id": "r1", "name": "A", "poly": [[0, 0], [1, 0], [1, 1]]})
|
|
v.ROOM_SCHEMA({"id": "r2", "name": "B", "x": 0, "y": 0, "w": 1, "h": 1})
|
|
with pytest.raises(vol.Invalid):
|
|
v.ROOM_SCHEMA({"id": "r3", "name": "C"})
|
|
with pytest.raises(vol.Invalid):
|
|
v.ROOM_SCHEMA({"id": "r4", "name": "D", "poly": [[0, 0], [1, 1]]})
|
|
|
|
|
|
def test_space_schema_drops_the_old_aspect_and_bounds_the_image_ratio():
|
|
"""v1.48.0: the canvas is square; only the IMAGE keeps proportions.
|
|
|
|
A stale tab may still send `aspect`. It is dropped rather than trusted —
|
|
the coordinates it arrives with were normalised against a different box, so
|
|
honouring the field would not make them right anyway.
|
|
"""
|
|
ok = {"id": "f1", "title": "1", "view_box": [0, 0, 1, 1], "rooms": []}
|
|
assert "aspect" not in v.SPACE_SCHEMA({**ok, "aspect": 1.4})
|
|
v.SPACE_SCHEMA({**ok, "plan_aspect": 1.4})
|
|
v.SPACE_SCHEMA({**ok, "plan_aspect": None})
|
|
with pytest.raises(vol.Invalid):
|
|
v.SPACE_SCHEMA({**ok, "plan_aspect": 0})
|
|
with pytest.raises(vol.Invalid):
|
|
v.SPACE_SCHEMA({**ok, "view_box": [0, 0, 1]})
|
|
|
|
|
|
def test_marker_schema():
|
|
v.MARKER_SCHEMA({"id": "m1", "binding": "device:abc"})
|
|
v.MARKER_SCHEMA({"id": "m2", "binding": "virtual", "name": "X",
|
|
"pdfs": [{"name": "a.pdf", "url": "/u/a.pdf"}]})
|
|
with pytest.raises(vol.Invalid):
|
|
v.MARKER_SCHEMA({"binding": "virtual"})
|
|
|
|
|
|
def test_config_schema_defaults_and_extra():
|
|
out = v.CONFIG_SCHEMA({"spaces": []})
|
|
assert out["markers"] == [] and out["settings"] == {}
|
|
out2 = v.CONFIG_SCHEMA({"spaces": [], "virtual_devices": [], "device_overrides": {}})
|
|
assert "spaces" in out2
|
|
|
|
|
|
def test_config_schema_full_roundtrip():
|
|
cfg = {
|
|
"spaces": [{
|
|
"id": "f1", "title": "Floor 1", "plan_url": "/p/f1.svg",
|
|
"aspect": 0.8, "view_box": [0, 0, 1, 1],
|
|
"rooms": [{"id": "r1", "name": "Hall", "area": "hall",
|
|
"poly": [[0, 0], [0.5, 0], [0.5, 0.5], [0, 0.5]]}],
|
|
"segments": [[0, 0, 0.5, 0]],
|
|
}],
|
|
"markers": [{"id": "d1", "binding": "device:x", "model": "M", "link": "https://e.com"}],
|
|
"settings": {"group_lights": True},
|
|
}
|
|
out = v.CONFIG_SCHEMA(cfg)
|
|
assert out["spaces"][0]["rooms"][0]["area"] == "hall"
|
|
assert out["markers"][0]["binding"] == "device:x"
|
|
|
|
|
|
def test_layout_schema():
|
|
v.LAYOUT_SCHEMA({"dev1": {"x": 0.1, "y": 0.2, "s": "f1"}})
|
|
with pytest.raises(vol.Invalid):
|
|
v.LAYOUT_SCHEMA({"dev1": {"x": 0.1}})
|
|
|
|
|
|
def test_space_display_settings():
|
|
"""Per-space display settings validate; garbage color/mode is rejected."""
|
|
ok = {
|
|
"id": "f1", "title": "Floor 1", "aspect": 1.0, "view_box": [0, 0, 1, 1],
|
|
"rooms": [], "settings": {
|
|
"show_borders": True, "show_names": False,
|
|
"room_color": "#3ea6ff", "room_opacity": 0.5, "fill_mode": "lqi",
|
|
},
|
|
}
|
|
v.SPACE_SCHEMA(ok)
|
|
import pytest as _pytest
|
|
bad_color = dict(ok, settings={"room_color": "javascript:x"})
|
|
with _pytest.raises(Exception):
|
|
v.SPACE_SCHEMA(bad_color)
|
|
bad_mode = dict(ok, settings={"fill_mode": "rainbow"})
|
|
with _pytest.raises(Exception):
|
|
v.SPACE_SCHEMA(bad_mode)
|
|
|
|
|
|
def test_space_temp_bounds():
|
|
"""Temperature comfort bounds validate as floats; temp fill mode accepted."""
|
|
ok = {
|
|
"id": "f1", "title": "F", "aspect": 1.0, "view_box": [0, 0, 1, 1], "rooms": [],
|
|
"settings": {"fill_mode": "temp", "temp_min": 19.5, "temp_max": "24"},
|
|
}
|
|
v.SPACE_SCHEMA(ok)
|
|
|
|
|
|
def test_finite_coordinates_rejected():
|
|
"""audit B5: NaN/Infinity coordinates must not reach storage."""
|
|
for bad in ("NaN", "Infinity", "-Infinity", float("nan"), float("inf")):
|
|
with pytest.raises(vol.Invalid):
|
|
v.LAYOUT_SCHEMA({"dev1": {"x": bad, "y": 0.5}})
|
|
assert v.LAYOUT_SCHEMA({"dev1": {"x": 0.5, "y": 0.25}})
|
|
|
|
|
|
def test_collection_caps():
|
|
"""audit B5: unbounded collections are capped."""
|
|
big = {f"d{i}": {"x": 0.1, "y": 0.1} for i in range(v.MAX_LAYOUT + 1)}
|
|
with pytest.raises(vol.Invalid):
|
|
v.LAYOUT_SCHEMA(big)
|
|
|
|
|
|
def test_finite_on_every_coordinate():
|
|
"""audit follow-up B5: NaN/Infinity must be refused everywhere, not only in layout."""
|
|
base = {"id": "s1", "title": "S", "aspect": 1.0, "view_box": [0, 0, 1, 1], "rooms": []}
|
|
# view_box
|
|
with pytest.raises(vol.Invalid):
|
|
v.CONFIG_SCHEMA({"spaces": [{**base, "view_box": [0, 0, "NaN", 100]}]})
|
|
# room rect coordinates
|
|
with pytest.raises(vol.Invalid):
|
|
v.CONFIG_SCHEMA({"spaces": [{**base, "rooms": [
|
|
{"id": "r", "name": "R", "x": "Infinity", "y": 0, "w": 1, "h": 1}]}]})
|
|
# polygon vertices
|
|
with pytest.raises(vol.Invalid):
|
|
v.CONFIG_SCHEMA({"spaces": [{**base, "rooms": [
|
|
{"id": "r", "name": "R", "poly": [[0, 0], [1, "NaN"], [1, 1]]}]}]})
|
|
# opening coordinates
|
|
with pytest.raises(vol.Invalid):
|
|
v.CONFIG_SCHEMA({"spaces": [{**base, "openings": [
|
|
{"id": "o", "type": "door", "x": "NaN", "y": 0.5, "angle": 0, "length": 0.1}]}]})
|
|
# a sane config still validates
|
|
assert v.CONFIG_SCHEMA({"spaces": [{**base, "rooms": [
|
|
{"id": "r", "name": "R", "poly": [[0, 0], [1, 0], [1, 1]]}]}]})
|
|
|
|
|
|
def test_geometry_magnitudes_are_bounded():
|
|
"""HP-1501-01: any finite float used to pass, and one schema-valid 1e100
|
|
room vertex made the space unviewable for every client — the exact failure
|
|
HP-1500-03 closed for layout positions, one schema over. ±4 is slack for a
|
|
vertex nudged past an edge, not an envelope for absurdity."""
|
|
base = {"id": "s1", "title": "S", "view_box": [0, 0, 1, 1], "rooms": []}
|
|
huge = 1e100
|
|
for cfg in (
|
|
{**base, "rooms": [{"id": "r", "name": "R", "poly": [[0, 0], [huge, 0], [1, 1]]}]},
|
|
{**base, "rooms": [{"id": "r", "name": "R", "poly": [[0, 0], [-huge, 0], [1, 1]]}]},
|
|
{**base, "rooms": [{"id": "r", "name": "R", "x": huge, "y": 0, "w": 1, "h": 1}]},
|
|
{**base, "rooms": [{"id": "r", "name": "R", "x": 0, "y": 0, "w": huge, "h": 1}]},
|
|
{**base, "view_box": [0, 0, huge, 1]},
|
|
{**base, "openings": [{"id": "o", "type": "door", "x": huge, "y": 0.5,
|
|
"angle": 0, "length": 0.1}]},
|
|
{**base, "openings": [{"id": "o", "type": "door", "x": 0.5, "y": 0.5,
|
|
"angle": 1e6, "length": 0.1}]},
|
|
):
|
|
with pytest.raises(vol.Invalid):
|
|
v.CONFIG_SCHEMA({"spaces": [cfg]})
|
|
# a vertex a bit past the canvas edge is a drawing, not an attack
|
|
assert v.CONFIG_SCHEMA({"spaces": [{**base, "rooms": [
|
|
{"id": "r", "name": "R", "poly": [[-0.2, 0], [1.3, 0], [1, 1]]}]}]})
|
|
|
|
|
|
def test_sizes_are_not_coordinates():
|
|
"""HP-1502-01: a size must be strictly positive — SVG refuses zero and
|
|
negative width/height, and the clients divide by these. view_box [0,0,0,0]
|
|
used to pass the shared validator and blank the plan on every client.
|
|
Coordinates may still be negative: a crop origin can sit past the edge."""
|
|
base = {"id": "s1", "title": "S", "view_box": [0, 0, 1, 1], "rooms": []}
|
|
for cfg in (
|
|
{**base, "view_box": [0, 0, 0, 0]},
|
|
{**base, "view_box": [0, 0, -1, -2]},
|
|
{**base, "view_box": [0, 0, 1, 0.0001]}, # below the 0.001 floor
|
|
{**base, "rooms": [{"id": "r", "name": "R", "x": 0.1, "y": 0.1, "w": 0, "h": 0.5}]},
|
|
{**base, "rooms": [{"id": "r", "name": "R", "x": 0.1, "y": 0.1, "w": 0.5, "h": -1}]},
|
|
):
|
|
with pytest.raises(vol.Invalid):
|
|
v.CONFIG_SCHEMA({"spaces": [cfg]})
|
|
# negative COORDINATES stay legal, and a normal crop viewport passes
|
|
assert v.CONFIG_SCHEMA({"spaces": [{**base, "view_box": [-0.2, -0.1, 1.4, 1.2]}]})
|
|
assert v.CONFIG_SCHEMA({"spaces": [{**base, "rooms": [
|
|
{"id": "r", "name": "R", "x": -0.1, "y": -0.1, "w": 0.4, "h": 0.3}]}]})
|
|
|
|
|
|
def test_openings_cap_enforced():
|
|
"""audit follow-up B5: MAX_OPENINGS was defined but never wired in."""
|
|
many = [{"id": f"o{i}", "type": "door", "x": 0.1, "y": 0.1, "angle": 0, "length": 0.1}
|
|
for i in range(v.MAX_OPENINGS + 1)]
|
|
with pytest.raises(vol.Invalid):
|
|
v.CONFIG_SCHEMA({"spaces": [{"id": "s1", "title": "S", "aspect": 1.0,
|
|
"view_box": [0, 0, 1, 1], "rooms": [],
|
|
"openings": many}]})
|
|
|
|
|
|
# ---------- plan-file collection (review R3-1) ----------
|
|
|
|
|
|
def _plans(tmp_path, names, age=0.0):
|
|
import os, time
|
|
d = tmp_path / "plans"
|
|
d.mkdir(exist_ok=True)
|
|
for n in names:
|
|
(d / n).write_bytes(b"x")
|
|
if age:
|
|
t = time.time() - age
|
|
os.utime(d / n, (t, t))
|
|
return d
|
|
|
|
|
|
def _cfg(*urls):
|
|
return {"spaces": [{"id": f"s{i}", "plan_url": u} for i, u in enumerate(urls)]}
|
|
|
|
|
|
def test_plan_basename_and_refs():
|
|
plan_basename, plan_refs, is_plan_file = plans.plan_basename, plans.plan_refs, plans.is_plan_file
|
|
|
|
assert plan_basename("/api/houseplan/content/plans/_/f1.abc.png?v=7") == "f1.abc.png"
|
|
assert plan_basename("/houseplan_files/plans/f1.svg") == "f1.svg"
|
|
assert plan_basename(None) == "" and plan_basename("") == "" and plan_basename(7) == ""
|
|
assert plan_refs(None) == set() and plan_refs({}) == set()
|
|
assert plan_refs(_cfg("/p/a.png", None, "/p/b.svg")) == {"a.png", "b.svg"}
|
|
assert is_plan_file("f1.svg") and is_plan_file("f1.tok.png")
|
|
assert not is_plan_file("notes.txt") and not is_plan_file("readme")
|
|
assert not is_plan_file("deep.name.with.dots.png") # 4 parts: not ours
|
|
|
|
|
|
def test_collect_plans_removes_only_the_superseded_file(tmp_path):
|
|
collect_plans = plans.collect_plans
|
|
|
|
d = _plans(tmp_path, ["f1.old.png", "f1.new.png", "f2.png"])
|
|
removed = collect_plans(d, _cfg("/p/f1.old.png", "/p/f2.png"), _cfg("/p/f1.new.png", "/p/f2.png"))
|
|
assert removed == 1
|
|
assert not (d / "f1.old.png").exists()
|
|
assert (d / "f1.new.png").is_file() and (d / "f2.png").is_file()
|
|
|
|
|
|
def test_collect_plans_keeps_a_fresh_unreferenced_upload(tmp_path):
|
|
"""Another client may be mid-transaction: its file is unreferenced but young."""
|
|
collect_plans = plans.collect_plans
|
|
|
|
d = _plans(tmp_path, ["f1.committed.png", "f1.inflight.png"])
|
|
removed = collect_plans(d, _cfg("/p/f1.committed.png"), _cfg("/p/f1.committed.png"))
|
|
assert removed == 0
|
|
assert (d / "f1.inflight.png").is_file()
|
|
|
|
|
|
def test_collect_plans_never_touches_a_referenced_or_foreign_file(tmp_path):
|
|
PLAN_ORPHAN_TTL_S = const.PLAN_ORPHAN_TTL_S
|
|
collect_plans = plans.collect_plans
|
|
|
|
old = PLAN_ORPHAN_TTL_S + 60
|
|
d = _plans(tmp_path, ["f1.png", "notes.txt", "readme", "deep.name.with.dots.png"], age=old)
|
|
removed = collect_plans(d, _cfg("/p/f1.png"), _cfg("/p/f1.png"))
|
|
assert removed == 0
|
|
for n in ("f1.png", "notes.txt", "readme", "deep.name.with.dots.png"):
|
|
assert (d / n).is_file()
|
|
|
|
|
|
def test_collect_plans_survives_a_missing_directory(tmp_path):
|
|
collect_plans = plans.collect_plans
|
|
|
|
assert collect_plans(tmp_path / "nope", _cfg(), _cfg()) == 0
|
|
|
|
|
|
def test_collect_plans_never_raises_when_the_directory_disappears(tmp_path, monkeypatch):
|
|
"""review R4-1: it runs behind a durable commit, so it may only report 0."""
|
|
collect_plans = plans.collect_plans
|
|
d = tmp_path / "plans"
|
|
d.mkdir()
|
|
|
|
def _boom(self):
|
|
raise OSError("gone")
|
|
|
|
monkeypatch.setattr(type(d), "iterdir", _boom, raising=False)
|
|
assert collect_plans(d, _cfg("/p/a.png"), _cfg("/p/b.png")) == 0
|
|
|
|
|
|
# ---------- the editor's options must be storable (issue #3) ----------
|
|
|
|
|
|
def _ts_list(name):
|
|
"""Read one `export const NAME = [...] as const;` list out of src/logic.ts.
|
|
|
|
Deliberately reads the TypeScript source rather than duplicating the values:
|
|
a list that lives in two places drifts, which is exactly what happened here.
|
|
"""
|
|
import re
|
|
|
|
src = os.path.join(os.path.dirname(os.path.dirname(__file__)), "src", "logic.ts")
|
|
with open(src, encoding="utf-8") as fh:
|
|
text = fh.read()
|
|
m = re.search(rf"export const {name} = \[(.*?)\] as const;", text, re.S)
|
|
assert m, f"{name} not found in src/logic.ts"
|
|
return re.findall(r"'([^']+)'", m.group(1))
|
|
|
|
|
|
def _marker(**extra):
|
|
return {"id": "m1", "binding": "entity:sensor.x", **extra}
|
|
|
|
|
|
def test_every_display_mode_the_editor_offers_is_accepted():
|
|
"""issue #3: 'value' was added to the card in v1.26.0 and never to the schema.
|
|
|
|
Saving a sensor set to "value instead of an icon" failed with
|
|
"not a valid value for dictionary value @ data['config']['markers'][n]['display']",
|
|
and because one bad marker rejects the whole config, the user could not save
|
|
at all. Reported 2026-07-27.
|
|
"""
|
|
modes = _ts_list("DISPLAY_MODES")
|
|
assert "value" in modes, "the regression this test exists for"
|
|
for mode in modes:
|
|
v.MARKER_SCHEMA(_marker(display=mode))
|
|
v.MARKER_SCHEMA(_marker(display=None))
|
|
with pytest.raises(vol.Invalid):
|
|
v.MARKER_SCHEMA(_marker(display="wat"))
|
|
|
|
|
|
def test_run_target_is_bounded_to_runnable_domains():
|
|
"""Owner's spec 2026-07-29: a tap may RUN automations, scripts and scenes —
|
|
and nothing else. The target rides the marker, so the schema is the door."""
|
|
for ok in ("automation.morning", "script.curtains", "scene.movie"):
|
|
v.MARKER_SCHEMA(_marker(tap_action="run", tap_target=ok, tap_confirm=True))
|
|
for bad in ("light.lamp", "lock.front_door", "shell_command.rm", "automation.", "x"):
|
|
with pytest.raises(vol.Invalid):
|
|
v.MARKER_SCHEMA(_marker(tap_action="run", tap_target=bad))
|
|
|
|
|
|
def test_every_tap_action_the_editor_offers_is_accepted():
|
|
for action in _ts_list("TAP_ACTIONS"):
|
|
v.MARKER_SCHEMA(_marker(tap_action=action))
|
|
with pytest.raises(vol.Invalid):
|
|
v.MARKER_SCHEMA(_marker(tap_action="launch-missiles"))
|
|
|
|
|
|
def _space(**settings):
|
|
return {
|
|
"id": "f1", "title": "F1", "aspect": 1.4, "view_box": [0, 0, 1, 1],
|
|
"rooms": [], "settings": settings,
|
|
}
|
|
|
|
|
|
def test_every_fill_mode_the_editor_offers_is_accepted():
|
|
for mode in _ts_list("SPACE_FILL_MODES"):
|
|
v.SPACE_SCHEMA(_space(fill_mode=mode))
|
|
with pytest.raises(vol.Invalid):
|
|
v.SPACE_SCHEMA(_space(fill_mode="rainbow"))
|
|
|
|
|
|
def test_every_room_fill_mode_the_editor_offers_is_accepted():
|
|
def room(mode):
|
|
return {
|
|
"id": "f1", "title": "F1", "aspect": 1.4, "view_box": [0, 0, 1, 1],
|
|
"rooms": [{"id": "r1", "name": "R", "x": 0.1, "y": 0.1, "w": 0.2, "h": 0.2,
|
|
"settings": {"fill_mode": mode}}],
|
|
}
|
|
|
|
for mode in _ts_list("ROOM_FILL_MODES"):
|
|
v.SPACE_SCHEMA(room(mode))
|
|
v.SPACE_SCHEMA(room(None)) # inherit from the space
|
|
with pytest.raises(vol.Invalid):
|
|
v.SPACE_SCHEMA(room("rainbow"))
|
|
|
|
|
|
# ---------- attachments & inner limits (HP-1454-02, -05) ----------
|
|
|
|
|
|
def test_reserve_filename_claims_the_name_atomically(tmp_path):
|
|
"""HP-1460-01: choosing a name and taking it must be one operation.
|
|
|
|
The old helper asked `exists()` and returned a string; two uploads racing
|
|
between the check and the write agreed on the same name and one overwrote
|
|
the other, both reporting success.
|
|
"""
|
|
reserve = plans.reserve_filename
|
|
d = tmp_path / "m1"
|
|
|
|
first = reserve(d, "manual.pdf")
|
|
assert first == "manual.pdf"
|
|
assert (d / first).is_file(), "the name is taken, not merely picked"
|
|
second = reserve(d, "manual.pdf")
|
|
assert second == "manual-2.pdf" and (d / second).is_file()
|
|
assert reserve(d, "manual.pdf") == "manual-3.pdf"
|
|
|
|
assert reserve(d, "readme") == "readme"
|
|
assert reserve(d, "readme") == "readme-2"
|
|
assert reserve(d, "../../etc/passwd") == "passwd"
|
|
|
|
|
|
def test_reserve_filename_result_survives_the_content_sanitizer(tmp_path):
|
|
"""A name the view would rewrite is a file written and then never served."""
|
|
reserve = plans.reserve_filename
|
|
d = tmp_path / "m2"
|
|
long_stem = "x" * 200 # far past the limit
|
|
names = [reserve(d, f"{long_stem}.pdf") for _ in range(12)]
|
|
assert len(set(names)) == 12, "each call takes its own name"
|
|
for n in names:
|
|
assert len(n) <= v.MAX_FILENAME
|
|
assert v.sanitize_filename(n) == n, n
|
|
assert n.endswith(".pdf")
|
|
|
|
# a name already exactly at the limit still leaves room for the tag
|
|
exact = "y" * (v.MAX_FILENAME - 4) + ".pdf"
|
|
assert len(exact) == v.MAX_FILENAME
|
|
a = reserve(d, exact)
|
|
b = reserve(d, exact)
|
|
assert a != b and len(b) <= v.MAX_FILENAME and v.sanitize_filename(b) == b
|
|
|
|
|
|
def test_reserve_filename_is_safe_under_concurrency(tmp_path):
|
|
"""Twenty threads, one filename: twenty distinct files, nothing overwritten."""
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
reserve = plans.reserve_filename
|
|
d = tmp_path / "m3"
|
|
d.mkdir(parents=True)
|
|
with ThreadPoolExecutor(max_workers=20) as pool:
|
|
names = list(pool.map(lambda _: reserve(d, "manual.pdf"), range(20)))
|
|
assert len(set(names)) == 20
|
|
assert sorted(p.name for p in d.iterdir()) == sorted(names)
|
|
|
|
|
|
def test_sweep_upload_temps(tmp_path):
|
|
"""HP-1460-02: a crashed transfer leaves a temp no other collector sees."""
|
|
import os
|
|
import time
|
|
|
|
files = tmp_path / "files"
|
|
files.mkdir()
|
|
fresh = files / f"{plans.TMP_PREFIX}fresh"
|
|
old = files / f"{plans.TMP_PREFIX}old"
|
|
keep = files / "notes.txt"
|
|
for f in (fresh, old, keep):
|
|
f.write_bytes(b"x")
|
|
t = time.time() - const.PLAN_ORPHAN_TTL_S - 60
|
|
os.utime(old, (t, t))
|
|
|
|
assert plans.sweep_upload_temps(files) == 1
|
|
assert not old.exists(), "an aged temporary goes"
|
|
assert fresh.is_file(), "a fresh one may belong to a request in flight"
|
|
assert keep.is_file(), "nothing else is touched"
|
|
assert plans.sweep_upload_temps(tmp_path / "nope") == 0
|
|
|
|
|
|
def _acfg(*pairs):
|
|
return {"markers": [{"id": f"m{i}", "pdfs": [{"url": f"/api/houseplan/content/files/{p}"}]}
|
|
for i, p in enumerate(pairs)]}
|
|
|
|
|
|
def test_attachment_refs_reads_marker_urls():
|
|
attachment_refs = plans.attachment_refs
|
|
assert attachment_refs(None) == set()
|
|
assert attachment_refs(_acfg("m1/a.pdf", "m2/b.pdf")) == {"m1/a.pdf", "m2/b.pdf"}
|
|
# legacy and foreign urls are not ours to collect against
|
|
cfg = {"markers": [{"id": "m", "pdfs": [{"url": "/local/x.pdf"}, {"url": "/api/houseplan/content/files/deep/a/b.pdf"}]}]}
|
|
assert plans.attachment_refs(cfg) == set()
|
|
|
|
|
|
def test_collect_attachments_removes_the_empty_folder_and_never_raises(tmp_path):
|
|
import os
|
|
import time
|
|
|
|
files = tmp_path / "files"
|
|
(files / "up_x").mkdir(parents=True)
|
|
f = files / "up_x" / "orphan.pdf"
|
|
f.write_bytes(b"x")
|
|
old = time.time() - const.PLAN_ORPHAN_TTL_S - 60
|
|
os.utime(f, (old, old))
|
|
assert plans.collect_attachments(files, {}, {}) == 1
|
|
assert not (files / "up_x").exists(), "the staging folder goes with its last file"
|
|
assert plans.collect_attachments(tmp_path / "nope", {}, {}) == 0
|
|
|
|
|
|
def test_inner_collection_limits():
|
|
room = {"id": "r", "name": "R", "poly": [[0.1, 0.1]] * v.MAX_POLY_POINTS}
|
|
v.ROOM_SCHEMA(room)
|
|
with pytest.raises(vol.Invalid):
|
|
v.ROOM_SCHEMA({**room, "poly": [[0.1, 0.1]] * (v.MAX_POLY_POINTS + 1)})
|
|
|
|
rect = {"id": "r", "name": "R", "x": 0.1, "y": 0.1, "w": 0.2, "h": 0.2}
|
|
v.ROOM_SCHEMA({**rect, "open_to": ["x"] * v.MAX_OPEN_TO})
|
|
with pytest.raises(vol.Invalid):
|
|
v.ROOM_SCHEMA({**rect, "open_to": ["x"] * (v.MAX_OPEN_TO + 1)})
|
|
|
|
m = {"id": "m", "binding": "virtual"}
|
|
v.MARKER_SCHEMA({**m, "controls": ["light.x"] * v.MAX_CONTROLS})
|
|
with pytest.raises(vol.Invalid):
|
|
v.MARKER_SCHEMA({**m, "controls": ["light.x"] * (v.MAX_CONTROLS + 1)})
|
|
|
|
pdf = {"name": "n", "url": "/api/houseplan/content/files/m/a.pdf"}
|
|
v.MARKER_SCHEMA({**m, "pdfs": [pdf] * v.MAX_PDFS})
|
|
with pytest.raises(vol.Invalid):
|
|
v.MARKER_SCHEMA({**m, "pdfs": [pdf] * (v.MAX_PDFS + 1)})
|
|
|
|
v.MARKER_SCHEMA({**m, "name": "n" * v.MAX_TEXT})
|
|
with pytest.raises(vol.Invalid):
|
|
v.MARKER_SCHEMA({**m, "name": "n" * (v.MAX_TEXT + 1)})
|
|
with pytest.raises(vol.Invalid):
|
|
v.MARKER_SCHEMA({**m, "link": "u" * (v.MAX_URL + 1)})
|
|
|
|
|
|
def test_legacy_segments_are_dropped_by_the_server():
|
|
"""A limit that depends on the client stripping the field is not a limit."""
|
|
out = v.SPACE_SCHEMA({
|
|
"id": "f1", "title": "F", "aspect": 1.4, "view_box": [0, 0, 1, 1], "rooms": [],
|
|
"segments": [[1, 2, 3, 4]] * 100000,
|
|
})
|
|
assert "segments" not in out
|
|
|
|
|
|
def _aged(path, seconds):
|
|
import os
|
|
import time
|
|
|
|
t = time.time() - seconds
|
|
os.utime(path, (t, t))
|
|
|
|
|
|
def _sp(sid, url):
|
|
return {"id": sid, "plan_url": f"/api/houseplan/content/plans/_/{url}" if url else None}
|
|
|
|
|
|
def test_plan_collection_matrix(tmp_path):
|
|
"""Which config transition means "the user asked for this file to go"?
|
|
|
|
`old_refs - new_refs` cannot tell replace, detach and delete-space apart —
|
|
they look identical. v1.46.4/v1.46.5 added guards for detach and only ever
|
|
reached them on the scheduled pass, so the commit itself still deleted a
|
|
detached plan the moment it was detached (HP-1465-01). One case is a
|
|
deletion the user asked for; the rest are kept.
|
|
"""
|
|
collect = plans.collect_plans
|
|
d = tmp_path / "plans"
|
|
d.mkdir()
|
|
|
|
def seed(*names):
|
|
for n in names:
|
|
(d / n).write_bytes(b"x")
|
|
_aged(d / n, const.SCHEDULED_GRACE_S * 2) # old enough for any rule
|
|
|
|
# 1. replace: the user picked a different image for the same space
|
|
seed("f1.old.png", "f1.new.png")
|
|
assert collect(d, {"spaces": [_sp("f1", "f1.old.png")]},
|
|
{"spaces": [_sp("f1", "f1.new.png")]}) == 1
|
|
assert not (d / "f1.old.png").exists() and (d / "f1.new.png").is_file()
|
|
|
|
# 2. detach: same space, switched to "draw"
|
|
seed("f2.png")
|
|
assert collect(d, {"spaces": [_sp("f2", "f2.png")]},
|
|
{"spaces": [_sp("f2", None)]}) == 0
|
|
assert (d / "f2.png").is_file(), "the editor says the file stays — it stays"
|
|
|
|
# 3. the space is deleted outright
|
|
seed("f3.png")
|
|
assert collect(d, {"spaces": [_sp("f3", "f3.png")]}, {"spaces": []}) == 0
|
|
assert (d / "f3.png").is_file()
|
|
|
|
# 4. the scheduled pass, later, still keeps both
|
|
cfg = {"spaces": [_sp("f2", None)]}
|
|
assert collect(d, cfg, cfg) == 0
|
|
assert (d / "f2.png").is_file() and (d / "f3.png").is_file()
|
|
|
|
# 5. an upload whose save was rejected is kept too, at any age. Ageing those
|
|
# out raced the retry: the sweep deleted a file the save was committing a
|
|
# reference to (caught by test_sweep_and_a_config_write_do_not_race).
|
|
seed("f4.current.png", "f4.reject.png")
|
|
live = {"spaces": [_sp("f4", "f4.current.png")]}
|
|
assert collect(d, live, live) == 0
|
|
assert (d / "f4.current.png").is_file() and (d / "f4.reject.png").is_file()
|
|
|
|
# 6. the same file still referenced by another space is never touched
|
|
seed("shared.png")
|
|
assert collect(d, {"spaces": [_sp("a", "shared.png"), _sp("b", "shared.png")]},
|
|
{"spaces": [_sp("a", None), _sp("b", "shared.png")]}) == 0
|
|
assert (d / "shared.png").is_file()
|
|
|
|
|
|
def test_attachment_collection_matrix(tmp_path):
|
|
"""Removing an attachment is a trash button; deleting the device is not."""
|
|
collect = plans.collect_attachments
|
|
|
|
def case(name):
|
|
d = tmp_path / name
|
|
d.mkdir()
|
|
return d
|
|
|
|
def seed(root, folder, fname, age=None):
|
|
(root / folder).mkdir(parents=True, exist_ok=True)
|
|
p = root / folder / fname
|
|
p.write_bytes(b"x")
|
|
if age:
|
|
_aged(p, age)
|
|
return p
|
|
|
|
def cfg(*markers):
|
|
return {"markers": [
|
|
{"id": mid, "pdfs": [{"url": f"/api/houseplan/content/files/{mid}/{n}"} for n in names]}
|
|
for mid, names in markers
|
|
]}
|
|
|
|
# 1. the user removed one attachment from a device that still exists
|
|
d = case("dropped")
|
|
seed(d, "m1", "dropped.pdf")
|
|
seed(d, "m1", "kept.pdf")
|
|
assert collect(d, cfg(("m1", ["dropped.pdf", "kept.pdf"])), cfg(("m1", ["kept.pdf"]))) == 1
|
|
assert not (d / "m1" / "dropped.pdf").exists()
|
|
assert (d / "m1" / "kept.pdf").is_file()
|
|
|
|
# 2. the device itself is gone: its manuals are not ours to throw away
|
|
d = case("device_gone")
|
|
seed(d, "m2", "manual.pdf", age=const.SCHEDULED_GRACE_S * 2)
|
|
assert collect(d, cfg(("m2", ["manual.pdf"])), cfg()) == 0
|
|
assert (d / "m2" / "manual.pdf").is_file()
|
|
# and the scheduled pass, later, agrees
|
|
assert collect(d, cfg(), cfg()) == 0
|
|
assert (d / "m2" / "manual.pdf").is_file()
|
|
|
|
# 3. a dialog that was never saved, in its own staging folder
|
|
d = case("staging")
|
|
seed(d, "up_x", "manual.pdf", age=const.PLAN_ORPHAN_TTL_S + 60)
|
|
assert collect(d, cfg(), cfg()) == 1
|
|
assert not (d / "up_x").exists()
|
|
|
|
# 4. an upload into a live device's folder whose save was rejected: kept,
|
|
# for the same reason as a plan's — a retry may be about to reference it
|
|
d = case("reject")
|
|
seed(d, "m3", "current.pdf")
|
|
seed(d, "m3", "rejected.pdf", age=const.SCHEDULED_GRACE_S + 60)
|
|
live = cfg(("m3", ["current.pdf"]))
|
|
assert collect(d, live, live) == 0
|
|
assert (d / "m3" / "current.pdf").is_file()
|
|
assert (d / "m3" / "rejected.pdf").is_file()
|
|
|
|
|
|
def test_only_a_staging_folder_ages_out(tmp_path):
|
|
"""The one age rule left. Everything else waits for the user to say so."""
|
|
import os
|
|
import time
|
|
|
|
files = tmp_path / "files"
|
|
(files / "m1").mkdir(parents=True)
|
|
(files / "up_abandoned").mkdir(parents=True)
|
|
ancient = time.time() - const.SCHEDULED_GRACE_S * 12
|
|
hour_ago = time.time() - const.PLAN_ORPHAN_TTL_S - 60
|
|
|
|
for path, when in (
|
|
((files / "m1" / "ancient.pdf"), ancient),
|
|
((files / "up_abandoned" / "manual.pdf"), hour_ago),
|
|
):
|
|
path.write_bytes(b"x")
|
|
os.utime(path, (when, when))
|
|
|
|
cfg = {"markers": [{"id": "m1", "pdfs": []}]}
|
|
assert plans.collect_attachments(files, cfg, cfg) == 1
|
|
assert (files / "m1" / "ancient.pdf").is_file(), "age alone is never a reason"
|
|
assert not (files / "up_abandoned").exists(), "a cancelled dialog goes after an hour"
|
|
|
|
|
|
|
|
|
|
# ---------- square canvas migration (v1.48.0) ----------
|
|
|
|
|
|
gm = _load_pure("geometry_migration")
|
|
|
|
|
|
def _sq(space, layout=None):
|
|
cfg = {"spaces": [space]}
|
|
gm.migrate_config(cfg, layout if layout is not None else {})
|
|
return cfg["spaces"][0]
|
|
|
|
|
|
def test_the_viewport_becomes_the_whole_square():
|
|
"""The grid is drawn over the view box, and the fit fits it.
|
|
|
|
Transforming the old rectangle instead would leave the new margins outside
|
|
the canvas — no grid there and nothing to draw on — which is the room the
|
|
square canvas was meant to add.
|
|
"""
|
|
sp = _sq({"id": "f1", "aspect": 0.5, "view_box": [0.1, 0.2, 0.5, 0.5], "rooms": []})
|
|
assert sp["view_box"] == [0.0, 0.0, 1.0, 1.0]
|
|
|
|
|
|
def test_a_wide_plan_gains_margins_above_and_below():
|
|
sp = _sq({
|
|
"id": "f1", "aspect": 2.0, "cell_cm": 5, "view_box": [0, 0, 1, 1],
|
|
"rooms": [{"id": "r", "x": 0.0, "y": 0.0, "w": 1.0, "h": 1.0}],
|
|
})
|
|
r = sp["rooms"][0]
|
|
assert (r["x"], r["w"]) == (0.0, 1.0), "the width is untouched"
|
|
assert r["y"] == 0.25 and r["h"] == 0.5, "half the height, centred"
|
|
assert sp["cell_cm"] == 5, "the grid is tied to the width, which did not change"
|
|
assert "aspect" not in sp
|
|
|
|
|
|
def test_a_tall_plan_gains_margins_on_the_sides_and_rescales_the_grid():
|
|
sp = _sq({
|
|
"id": "f1", "aspect": 0.5, "cell_cm": 5, "view_box": [0, 0, 1, 1],
|
|
"rooms": [{"id": "r", "poly": [[0, 0], [1, 0], [1, 1], [0, 1]]}],
|
|
})
|
|
poly = sp["rooms"][0]["poly"]
|
|
assert [round(c, 6) for c in poly[0]] == [0.25, 0.0]
|
|
assert [round(c, 6) for c in poly[2]] == [0.75, 1.0], "half the width, centred"
|
|
assert sp["cell_cm"] == 10, "the canvas got twice as wide, so a cell is twice the cm"
|
|
|
|
|
|
def test_a_square_plan_is_left_alone():
|
|
before = {
|
|
"id": "f1", "aspect": 1.0, "cell_cm": 5, "view_box": [0, 0, 1, 1],
|
|
"rooms": [{"id": "r", "x": 0.1, "y": 0.2, "w": 0.3, "h": 0.4}],
|
|
}
|
|
sp = _sq({**before, "rooms": [dict(before["rooms"][0])]})
|
|
assert sp["rooms"][0] == before["rooms"][0]
|
|
assert sp["cell_cm"] == 5 and sp["view_box"] == [0.0, 0.0, 1.0, 1.0]
|
|
|
|
|
|
def test_migration_preserves_real_lengths_and_shapes():
|
|
"""A wall keeps its length in centimetres, and a square stays square."""
|
|
GRID = 1000.0
|
|
|
|
def wall_cm(space, p, q):
|
|
# render units per normalised unit is the canvas width, always 1000
|
|
dx = (q[0] - p[0]) * GRID
|
|
dy = (q[1] - p[1]) * GRID
|
|
pitch = GRID / 40 # whatever the grid is, the same constant both sides
|
|
return ((dx * dx + dy * dy) ** 0.5 / pitch) * float(space["cell_cm"])
|
|
|
|
for aspect in (2.0, 0.5, 0.8155784250916674, 1.4142):
|
|
# a square room, 0.2 x 0.2 of the OLD box, i.e. 200 x 200/aspect render
|
|
old = {"id": "f", "aspect": aspect, "cell_cm": 5,
|
|
"rooms": [{"id": "r", "poly": [[0.2, 0.2], [0.4, 0.2], [0.4, 0.4], [0.2, 0.4]]}]}
|
|
before_w = 0.2 * GRID
|
|
before_h = 0.2 * GRID / aspect
|
|
before_cm_w = (before_w / (GRID / 40)) * 5
|
|
sp = _sq(old)
|
|
poly = sp["rooms"][0]["poly"]
|
|
after_w = (poly[1][0] - poly[0][0]) * GRID
|
|
after_h = (poly[2][1] - poly[1][1]) * GRID
|
|
assert abs(after_w / after_h - before_w / before_h) < 1e-9, "shape preserved"
|
|
# cell_cm is stored rounded — a user reads it — so allow 0.01 cm on a
|
|
# 40 cm wall rather than pretending the scale is infinitely precise
|
|
assert abs(wall_cm(sp, poly[0], poly[1]) - before_cm_w) < 1e-2, "length in cm preserved"
|
|
|
|
|
|
def test_migration_moves_marker_positions_of_that_space_only():
|
|
layout = {
|
|
"a": {"s": "f1", "x": 0.5, "y": 0.5},
|
|
"b": {"s": "other", "x": 0.5, "y": 0.5},
|
|
"c": "not a dict",
|
|
}
|
|
cfg = {"spaces": [{"id": "f1", "aspect": 2.0, "rooms": []},
|
|
{"id": "other", "rooms": []}]}
|
|
assert gm.migrate_config(cfg, layout) is True
|
|
assert layout["a"] == {"s": "f1", "x": 0.5, "y": 0.5}, "x untouched for a wide plan"
|
|
assert layout["a"]["y"] == 0.5
|
|
assert layout["b"] == {"s": "other", "x": 0.5, "y": 0.5}, "another space is not touched"
|
|
|
|
|
|
def test_migration_runs_once_and_only_when_needed():
|
|
cfg = {"spaces": [{"id": "f1", "aspect": 2.0, "rooms": [], "cell_cm": 5}]}
|
|
assert gm.migrate_config(cfg, {}) is True
|
|
snapshot = repr(cfg)
|
|
assert gm.migrate_config(cfg, {}) is False, "already square: nothing to do"
|
|
assert repr(cfg) == snapshot
|
|
|
|
|
|
def test_migration_survives_a_crash_between_the_two_store_writes():
|
|
"""HP-1490-01: the two stores are written independently and either write
|
|
can fail. The intent (space -> old aspect) is saved BEFORE anything moves
|
|
and cleared with the layout write, so whichever half is missing after a
|
|
crash, the next start finishes exactly it — once.
|
|
"""
|
|
cfg = {"spaces": [{"id": "f1", "aspect": 2.0, "rooms": []}]}
|
|
layout = {"m": {"s": "f1", "x": 0.1, "y": 0.1}}
|
|
|
|
# start of the migration: the intent is computed from the config
|
|
pending = gm.pending_from_config(cfg)
|
|
assert pending == {"f1": 2.0}
|
|
|
|
# the config half commits; the process dies before the layout half
|
|
assert gm.migrate_config(cfg) is True
|
|
assert gm.pending_from_config(cfg) == {}, "the trigger left with the config write"
|
|
|
|
# next start: the config offers nothing, the SAVED intent still knows
|
|
assert gm.migrate_layout(layout, pending) is True
|
|
assert layout["m"] == {"s": "f1", "x": 0.1, "y": 0.3}, "y is re-centred for a wide plan"
|
|
|
|
# and the layout half never runs twice, because the intent is cleared by
|
|
# the same write that stores the migrated layout — with no intent there is
|
|
# nothing to apply
|
|
assert gm.migrate_layout(layout, {}) is False
|
|
assert layout["m"] == {"s": "f1", "x": 0.1, "y": 0.3}
|
|
|
|
|
|
def test_migration_intent_is_the_union_of_saved_and_current():
|
|
"""A crash BEFORE the config write leaves both the intent and the aspects;
|
|
merging them must not double anything, and a space added to the config
|
|
since (there cannot be one mid-crash, but the code should not care) still
|
|
migrates."""
|
|
cfg = {"spaces": [{"id": "f1", "aspect": 2.0, "rooms": []}]}
|
|
saved = {"f1": 2.0}
|
|
merged = {**saved, **gm.pending_from_config(cfg)}
|
|
assert merged == {"f1": 2.0}
|
|
layout = {"m": {"s": "f1", "x": 0.1, "y": 0.1}}
|
|
gm.migrate_config(cfg)
|
|
gm.migrate_layout(layout, merged)
|
|
assert layout["m"]["y"] == 0.3
|
|
assert cfg["spaces"][0]["view_box"] == [0.0, 0.0, 1.0, 1.0]
|
|
|
|
|
|
# ---------- store-wide limits (HP-1470-01) ----------
|
|
|
|
|
|
def test_check_quota_counts_the_whole_store_not_one_request(tmp_path):
|
|
"""Per-request caps say nothing about how many requests there are."""
|
|
d = tmp_path / "plans"
|
|
d.mkdir()
|
|
for i in range(3):
|
|
(d / f"p{i}.png").write_bytes(b"x" * 1000)
|
|
|
|
plans.check_quota(d, 1000, max_bytes=10_000, max_files=10) # fits
|
|
|
|
with pytest.raises(plans.QuotaError) as e:
|
|
plans.check_quota(d, 8000, max_bytes=10_000, max_files=10)
|
|
assert e.value.reason == "quota_exceeded" and "MB" in e.value.detail
|
|
|
|
with pytest.raises(plans.QuotaError) as e:
|
|
plans.check_quota(d, 1, max_bytes=10_000, max_files=3)
|
|
assert e.value.reason == "too_many_files"
|
|
|
|
|
|
def test_dir_usage_walks_subfolders_and_ignores_the_unreadable(tmp_path):
|
|
d = tmp_path / "files"
|
|
(d / "m1").mkdir(parents=True)
|
|
(d / "m1" / "a.pdf").write_bytes(b"x" * 10)
|
|
(d / "b.pdf").write_bytes(b"x" * 5)
|
|
assert plans.dir_usage(d) == (15, 2)
|
|
assert plans.dir_usage(tmp_path / "nope") == (0, 0)
|
|
|
|
|
|
def test_check_quota_refuses_when_the_disk_is_nearly_full(tmp_path, monkeypatch):
|
|
import shutil
|
|
|
|
d = tmp_path / "plans"
|
|
d.mkdir()
|
|
monkeypatch.setattr(
|
|
shutil, "disk_usage", lambda _p: type("U", (), {"free": const.MIN_FREE_BYTES // 2})()
|
|
)
|
|
with pytest.raises(plans.QuotaError) as e:
|
|
plans.check_quota(d, 1, max_bytes=10 ** 12, max_files=10 ** 6)
|
|
assert e.value.reason == "low_disk_space"
|