fix: pre-coercion weekday entry validation, repr-keyed error sort, exact model names only, relative-import alias scan, strict int typing
- direct list construction validates entries before set() coercion
(bool collapse and unhashable items both raise RuleValidationError)
- unknown-key error sorts by repr (mixed-type keys never leak TypeError)
- string rule_type accepted ONLY as exact model names: storage aliases
('monthly'/'custom'), whitespace padding, and case variants rejected;
the DAO owns storage-string mapping
- purity guard checks ImportFrom alias names so 'from . import db' is
caught, not just node.module
- _is_plain_int requires type(value) is int: unhashable int subclasses
cannot construct
This commit is contained in:
@@ -105,8 +105,19 @@ def _validate_date(value: str, field_name: str) -> None:
|
|||||||
|
|
||||||
|
|
||||||
def _validate_weekday_set(weekday_set) -> frozenset[int]:
|
def _validate_weekday_set(weekday_set) -> frozenset[int]:
|
||||||
"""Return a normalized frozenset or raise with a clear message."""
|
"""Return a normalized frozenset or raise with a clear message.
|
||||||
|
|
||||||
|
Sequence entries are validated BEFORE set() coercion: a list like
|
||||||
|
[0, False] must not collapse the boolean into 0, and unhashable
|
||||||
|
entries must raise RuleValidationError, never a raw TypeError.
|
||||||
|
"""
|
||||||
if isinstance(weekday_set, (list, tuple)):
|
if isinstance(weekday_set, (list, tuple)):
|
||||||
|
for entry in weekday_set:
|
||||||
|
if isinstance(entry, bool) or not isinstance(entry, int):
|
||||||
|
raise RuleValidationError(
|
||||||
|
f"weekday_set entries must be plain integers, got "
|
||||||
|
f"{entry!r}"
|
||||||
|
)
|
||||||
weekday_set = set(weekday_set)
|
weekday_set = set(weekday_set)
|
||||||
if not isinstance(weekday_set, (set, frozenset)):
|
if not isinstance(weekday_set, (set, frozenset)):
|
||||||
raise RuleValidationError(
|
raise RuleValidationError(
|
||||||
@@ -132,7 +143,9 @@ def _validate_weekday_set(weekday_set) -> frozenset[int]:
|
|||||||
|
|
||||||
|
|
||||||
def _is_plain_int(value) -> bool:
|
def _is_plain_int(value) -> bool:
|
||||||
return isinstance(value, int) and not isinstance(value, bool)
|
"""True only for real ints: bool excluded AND int subclasses too
|
||||||
|
(an unhashable int subclass would break the frozen-hashable model)."""
|
||||||
|
return type(value) is int
|
||||||
|
|
||||||
|
|
||||||
def _validate_monthly_weekday(nth_weekday, weekday) -> None:
|
def _validate_monthly_weekday(nth_weekday, weekday) -> None:
|
||||||
@@ -182,40 +195,19 @@ class ScheduleRule:
|
|||||||
# Normalize a string rule_type into the enum; anything else
|
# Normalize a string rule_type into the enum; anything else
|
||||||
# that is not already a RuleType is rejected.
|
# that is not already a RuleType is rejected.
|
||||||
if isinstance(self.rule_type, str):
|
if isinstance(self.rule_type, str):
|
||||||
normalized = self.rule_type.strip().lower()
|
# EXACT model names only: no whitespace stripping, no storage
|
||||||
# Accept the model name (monthly_day) or the storage value
|
# aliases. The DAO maps storage strings to model fields when
|
||||||
# ('monthly'/'custom'), the latter resolved by the populated
|
# reading rows; the model boundary stays unambiguous.
|
||||||
# fields per the DAO mapping.
|
|
||||||
candidates = {rt.name.lower(): rt for rt in RuleType}
|
candidates = {rt.name.lower(): rt for rt in RuleType}
|
||||||
if normalized in candidates:
|
if self.rule_type in candidates:
|
||||||
object.__setattr__(self, "rule_type", candidates[normalized])
|
object.__setattr__(
|
||||||
|
self, "rule_type", candidates[self.rule_type]
|
||||||
|
)
|
||||||
else:
|
else:
|
||||||
by_storage = {
|
raise RuleValidationError(
|
||||||
rt.storage_value: rt for rt in RuleType
|
f"rule_type must be a RuleType or one of "
|
||||||
}
|
f"{sorted(candidates)}, got {self.rule_type!r}"
|
||||||
resolved = None
|
)
|
||||||
if normalized in by_storage:
|
|
||||||
candidate = by_storage[normalized]
|
|
||||||
# 'monthly' is ambiguous; resolve by fields.
|
|
||||||
if normalized == "monthly":
|
|
||||||
if self.nth_weekday is not None:
|
|
||||||
resolved = RuleType.MONTHLY_WEEKDAY
|
|
||||||
else:
|
|
||||||
resolved = RuleType.MONTHLY_DAY
|
|
||||||
elif normalized == "custom":
|
|
||||||
resolved = RuleType.CUSTOM_DAYS
|
|
||||||
elif normalized == "custom" and self.weekday_set is not None:
|
|
||||||
resolved = RuleType.CUSTOM_DAYS
|
|
||||||
if resolved is None:
|
|
||||||
resolved = candidates.get(self.rule_type)
|
|
||||||
if resolved is not None:
|
|
||||||
object.__setattr__(self, "rule_type", resolved)
|
|
||||||
else:
|
|
||||||
raise RuleValidationError(
|
|
||||||
f"rule_type must be one of "
|
|
||||||
f"{sorted(rt.name.lower() for rt in RuleType)} "
|
|
||||||
f"or a storage value, got {self.rule_type!r}"
|
|
||||||
)
|
|
||||||
if not isinstance(self.rule_type, RuleType):
|
if not isinstance(self.rule_type, RuleType):
|
||||||
raise RuleValidationError(
|
raise RuleValidationError(
|
||||||
f"rule_type must be a RuleType or one of its string "
|
f"rule_type must be a RuleType or one of its string "
|
||||||
@@ -365,7 +357,9 @@ class ScheduleRule:
|
|||||||
}
|
}
|
||||||
unknown = [key for key in data if key not in allowed]
|
unknown = [key for key in data if key not in allowed]
|
||||||
if unknown:
|
if unknown:
|
||||||
raise RuleValidationError(f"unknown rule fields: {sorted(unknown)}")
|
raise RuleValidationError(
|
||||||
|
f"unknown rule fields: {sorted(unknown, key=repr)}"
|
||||||
|
)
|
||||||
if "rule_type" not in data:
|
if "rule_type" not in data:
|
||||||
raise RuleValidationError("missing required rule field: rule_type")
|
raise RuleValidationError("missing required rule field: rule_type")
|
||||||
weekday_set = data.get("weekday_set")
|
weekday_set = data.get("weekday_set")
|
||||||
|
|||||||
@@ -297,19 +297,26 @@ def test_recurrence_module_imports_only_stdlib() -> None:
|
|||||||
elif isinstance(node, ast.ImportFrom):
|
elif isinstance(node, ast.ImportFrom):
|
||||||
module = (node.module or "").lower()
|
module = (node.module or "").lower()
|
||||||
level = node.level # 1+ means relative import
|
level = node.level # 1+ means relative import
|
||||||
# Relative imports into the integration package (db/dao
|
# 'from . import db' has module=None and the name in the
|
||||||
# clients) are forbidden; schema.py is also off-limits.
|
# aliases: check BOTH the module and every imported name.
|
||||||
|
imported_names = [alias.name for alias in node.names]
|
||||||
|
forbidden_targets = {
|
||||||
|
"db", "schema", "dao_children", "dao_rules",
|
||||||
|
"dao_presence", "dao_instances", "migrations", "store",
|
||||||
|
}
|
||||||
if level:
|
if level:
|
||||||
assert "db" not in module.split(".") and module not in {
|
for name in imported_names:
|
||||||
"db",
|
assert name not in forbidden_targets, (
|
||||||
"schema",
|
f"recurrence.py must not import {name!r} from a "
|
||||||
"dao_children",
|
"relative import"
|
||||||
"dao_rules",
|
)
|
||||||
"dao_presence",
|
if module:
|
||||||
"dao_instances",
|
assert module not in forbidden_targets, (
|
||||||
"migrations",
|
f"recurrence.py must not import from .{module}"
|
||||||
"store",
|
)
|
||||||
}, f"recurrence.py must not import from .{module}"
|
assert "db" not in module.split("."), (
|
||||||
|
f"recurrence.py must not import from {module}"
|
||||||
|
)
|
||||||
assert "homeassistant" not in module, (
|
assert "homeassistant" not in module, (
|
||||||
f"recurrence.py must not import from {module}"
|
f"recurrence.py must not import from {module}"
|
||||||
)
|
)
|
||||||
@@ -409,3 +416,50 @@ def test_rule_type_is_a_dataclass() -> None:
|
|||||||
import dataclasses
|
import dataclasses
|
||||||
|
|
||||||
assert dataclasses.is_dataclass(ScheduleRule)
|
assert dataclasses.is_dataclass(ScheduleRule)
|
||||||
|
|
||||||
|
def test_direct_list_with_bool_weekday_rejected(tmp_path) -> None:
|
||||||
|
"""Direct construction with a list: entries validated pre-coercion."""
|
||||||
|
with pytest.raises(RuleValidationError, match="plain integers"):
|
||||||
|
ScheduleRule(
|
||||||
|
rule_type=RuleType.WEEKLY, weekday_set=[0, False],
|
||||||
|
start_date="2026-09-01",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_unhashable_list_item_leads_to_clear_error(tmp_path) -> None:
|
||||||
|
"""Unhashable entries raise RuleValidationError, not TypeError."""
|
||||||
|
with pytest.raises(RuleValidationError):
|
||||||
|
ScheduleRule(
|
||||||
|
rule_type=RuleType.WEEKLY, weekday_set=[[0]],
|
||||||
|
start_date="2026-09-01",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_from_dict_mixed_type_unknown_keys_raise_validation_error(
|
||||||
|
tmp_path,
|
||||||
|
) -> None:
|
||||||
|
"""sorted() over mixed key types must not leak TypeError."""
|
||||||
|
with pytest.raises(RuleValidationError, match="unknown rule fields"):
|
||||||
|
ScheduleRule.from_dict({"rule_type": "daily", 42: "value"})
|
||||||
|
|
||||||
|
|
||||||
|
def test_storage_aliases_and_whitespace_are_not_model_names(tmp_path) -> None:
|
||||||
|
"""The model boundary is exact names only: storage strings, padded
|
||||||
|
and case variants are all rejected."""
|
||||||
|
for bad in ("monthly", "custom", " DAILY ", "daily ", "DAILY"):
|
||||||
|
with pytest.raises(RuleValidationError):
|
||||||
|
ScheduleRule(rule_type=bad, start_date="2026-09-01")
|
||||||
|
|
||||||
|
|
||||||
|
def test_unhashable_int_subclass_interval_rejected() -> None:
|
||||||
|
"""An int subclass with __hash__ = None must not construct."""
|
||||||
|
|
||||||
|
class UnhashableInt(int):
|
||||||
|
__hash__ = None
|
||||||
|
|
||||||
|
with pytest.raises(RuleValidationError):
|
||||||
|
ScheduleRule(
|
||||||
|
rule_type=RuleType.DAILY,
|
||||||
|
interval=UnhashableInt(1),
|
||||||
|
start_date="2026-09-01",
|
||||||
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user