mirror of
https://github.com/Matysh/houseplan-card
synced 2026-07-31 16:38:31 +00:00
Matysh
df65d25348
The ±4 bound from v1.50.2 measured view_box[2:4] and room w/h with the same ruler as coordinates, so zero and negative sizes still passed the schema — and viewBox='0 0 0 0' draws nothing on every client, with the static card computing aspect-ratio: 0 / 0 on top. _EXTENT now requires strictly positive sizes with a floor of one thousandth of the canvas (1 render unit — far below any real room, keeps the maths finite); coordinates stay allowed to be negative, a crop origin legitimately sits past the edge. Defensive layer for stores that already hold a broken viewport: spaceModels falls back to the whole canvas — both cards render from that model, so both get the fallback — and a legacy rectangle with a negative size reads as the same rectangle drawn from the other corner. Also: the room settings button is the bottom row of the room card, and the room name renders in the same spot in view and plan modes (owner's request, committed earlier on dev).
916 lines
36 KiB
Python
916 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_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"
|