feat(realism): scaffold decnet/realism/ library
Empty subpackage skeleton for the realism migration: ContentClass enum (file/email/canary content categories), Plan dataclass (frozen, with edit-action invariant), in_work_hours window check (wrap-around supported, fail-open on parse error), and sample_mtime for backdated file timestamps that snap into a persona's active hours. Stage 1 of the orchestrator+canary realism unification — no production caller wired yet; planner.pick is a stub returning None until stage 3.
This commit is contained in:
0
tests/realism/__init__.py
Normal file
0
tests/realism/__init__.py
Normal file
120
tests/realism/test_diurnal.py
Normal file
120
tests/realism/test_diurnal.py
Normal file
@@ -0,0 +1,120 @@
|
||||
"""Coverage for :mod:`decnet.realism.diurnal`.
|
||||
|
||||
Two functions to exercise:
|
||||
|
||||
* :func:`in_work_hours` — straightforward window membership including
|
||||
the wrap-around (``22:00-06:00``) case and the fail-open behaviour
|
||||
on malformed windows.
|
||||
* :func:`sample_mtime` — must (a) return a ``datetime`` strictly in
|
||||
the past, (b) clip to the configured backdate cap, and (c) snap the
|
||||
hour-of-day into the persona's window when the unconstrained
|
||||
candidate would land outside.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from decnet.realism.diurnal import in_work_hours, sample_mtime
|
||||
|
||||
|
||||
# Fixed 'now' for reproducible tests — Monday 2026-04-27 14:00 UTC.
|
||||
_NOW = datetime(2026, 4, 27, 14, 0, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
# ---- in_work_hours -----------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"now_hour,now_min,window,expected",
|
||||
[
|
||||
(10, 0, "09:00-18:00", True),
|
||||
(8, 59, "09:00-18:00", False),
|
||||
(9, 0, "09:00-18:00", True),
|
||||
(18, 0, "09:00-18:00", False), # exclusive end
|
||||
(17, 59, "09:00-18:00", True),
|
||||
(23, 30, "22:00-06:00", True), # wrap-around: late
|
||||
(3, 0, "22:00-06:00", True), # wrap-around: early
|
||||
(12, 0, "22:00-06:00", False), # wrap-around: middle of day
|
||||
],
|
||||
)
|
||||
def test_in_work_hours_window_membership(
|
||||
now_hour: int, now_min: int, window: str, expected: bool,
|
||||
) -> None:
|
||||
now = _NOW.replace(hour=now_hour, minute=now_min)
|
||||
assert in_work_hours(window, now) is expected
|
||||
|
||||
|
||||
def test_in_work_hours_equal_start_end_means_always_on() -> None:
|
||||
# A persona pegged "00:00-00:00" should never be silenced by the
|
||||
# diurnal gate — interpreted as "no schedule".
|
||||
assert in_work_hours("00:00-00:00", _NOW) is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"garbage",
|
||||
["not-a-window", "9-18", "09:00", "25:00-26:00", "09:00-18:99", ""],
|
||||
)
|
||||
def test_malformed_window_fails_open(garbage: str) -> None:
|
||||
# The fleet must not silence on a typo — same fail-open semantics
|
||||
# as decnet.orchestrator.emailgen.personas.in_active_hours.
|
||||
assert in_work_hours(garbage, _NOW) is True
|
||||
|
||||
|
||||
# ---- sample_mtime ------------------------------------------------------
|
||||
|
||||
def test_sample_mtime_is_in_the_past() -> None:
|
||||
rng = random.Random(0)
|
||||
for _ in range(20):
|
||||
mt = sample_mtime("09:00-18:00", _NOW, rand=rng)
|
||||
assert mt < _NOW, f"sample_mtime returned future: {mt} >= {_NOW}"
|
||||
|
||||
|
||||
def test_sample_mtime_respects_backdate_cap() -> None:
|
||||
rng = random.Random(0)
|
||||
cap_days = 7.0
|
||||
for _ in range(50):
|
||||
mt = sample_mtime(
|
||||
"09:00-18:00", _NOW, rand=rng,
|
||||
backdate_max_days=cap_days, backdate_min_hours=0.5,
|
||||
)
|
||||
assert _NOW - mt <= timedelta(days=cap_days) + timedelta(hours=1)
|
||||
assert _NOW - mt >= timedelta(hours=0.5) - timedelta(seconds=1)
|
||||
|
||||
|
||||
def test_sample_mtime_snaps_hour_into_window() -> None:
|
||||
# Force a tight window then assert the hour-of-day is always in it.
|
||||
rng = random.Random(42)
|
||||
window = "09:00-18:00"
|
||||
for _ in range(60):
|
||||
mt = sample_mtime(window, _NOW, rand=rng)
|
||||
assert 9 <= mt.hour < 18, (
|
||||
f"hour {mt.hour} fell outside {window} on {mt.isoformat()}"
|
||||
)
|
||||
|
||||
|
||||
def test_sample_mtime_handles_wrap_around_window() -> None:
|
||||
rng = random.Random(123)
|
||||
for _ in range(40):
|
||||
mt = sample_mtime("22:00-06:00", _NOW, rand=rng)
|
||||
assert mt.hour >= 22 or mt.hour < 6, (
|
||||
f"hour {mt.hour} fell outside wrap window on {mt.isoformat()}"
|
||||
)
|
||||
|
||||
|
||||
def test_sample_mtime_malformed_window_does_not_snap() -> None:
|
||||
# When the window can't be parsed, just return the unconstrained
|
||||
# backdate. Belt-and-braces: shouldn't crash, shouldn't future-stamp.
|
||||
rng = random.Random(0)
|
||||
mt = sample_mtime("garbage", _NOW, rand=rng)
|
||||
assert mt < _NOW
|
||||
|
||||
|
||||
def test_sample_mtime_is_deterministic_per_seed() -> None:
|
||||
# The diurnal sampler accepts a Random — pinning the seed must
|
||||
# produce stable output, otherwise tests can't assert anything
|
||||
# tighter than "returns a datetime in the past."
|
||||
a = sample_mtime("09:00-18:00", _NOW, rand=random.Random(7))
|
||||
b = sample_mtime("09:00-18:00", _NOW, rand=random.Random(7))
|
||||
assert a == b
|
||||
102
tests/realism/test_taxonomy.py
Normal file
102
tests/realism/test_taxonomy.py
Normal file
@@ -0,0 +1,102 @@
|
||||
"""Coverage for :mod:`decnet.realism.taxonomy`.
|
||||
|
||||
The enum values are persisted on ``synthetic_files.content_class`` and
|
||||
flow through bus topics — renaming a member is a schema change, so the
|
||||
stable-list test pins the wire format. ``Plan`` invariants (frozen,
|
||||
edit requires previous_body) are tested too because the planner relies
|
||||
on construction-time validation rather than a separate validator pass.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from decnet.realism.taxonomy import ContentClass, Plan
|
||||
|
||||
|
||||
def test_content_class_values_are_stable() -> None:
|
||||
# If anyone renames or reorders, the assertion explodes — the
|
||||
# enum is wire-visible (synthetic_files.content_class column,
|
||||
# bus event payloads) so changes need a schema bump elsewhere.
|
||||
assert {c.value for c in ContentClass} == {
|
||||
"note", "todo", "draft", "script",
|
||||
"log_cron", "log_daemon", "cache_tmp",
|
||||
"email",
|
||||
"canary_aws_creds", "canary_env_file", "canary_git_config",
|
||||
"canary_ssh_key", "canary_honeydoc", "canary_honeydoc_docx",
|
||||
"canary_honeydoc_pdf", "canary_mysql_dump",
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", ["NOTE", "TODO", "DRAFT", "SCRIPT"])
|
||||
def test_user_classes_classified(name: str) -> None:
|
||||
cls = ContentClass[name]
|
||||
assert cls.is_user_class()
|
||||
assert not cls.is_system_class()
|
||||
assert not cls.is_canary()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", ["LOG_CRON", "LOG_DAEMON", "CACHE_TMP"])
|
||||
def test_system_classes_classified(name: str) -> None:
|
||||
cls = ContentClass[name]
|
||||
assert cls.is_system_class()
|
||||
assert not cls.is_user_class()
|
||||
assert not cls.is_canary()
|
||||
|
||||
|
||||
def test_canary_members_all_classified() -> None:
|
||||
canaries = [c for c in ContentClass if c.value.startswith("canary_")]
|
||||
assert canaries, "expected at least one canary content_class"
|
||||
for c in canaries:
|
||||
assert c.is_canary()
|
||||
assert not c.is_user_class()
|
||||
assert not c.is_system_class()
|
||||
|
||||
|
||||
def test_email_is_neither_user_nor_system_nor_canary() -> None:
|
||||
# Email lives on its own track — same content engine but a
|
||||
# different driver and a different table. Classification helpers
|
||||
# must not falsely group it into file-class buckets.
|
||||
assert ContentClass.EMAIL.value == "email"
|
||||
assert not ContentClass.EMAIL.is_user_class()
|
||||
assert not ContentClass.EMAIL.is_system_class()
|
||||
assert not ContentClass.EMAIL.is_canary()
|
||||
|
||||
|
||||
def _plan(**kw):
|
||||
defaults = dict(
|
||||
decky_uuid="d-1",
|
||||
decky_name="alpha",
|
||||
persona="admin",
|
||||
content_class=ContentClass.NOTE,
|
||||
action="create",
|
||||
target_path="/home/admin/notes.txt",
|
||||
mtime=datetime(2026, 4, 25, 11, 30, tzinfo=timezone.utc),
|
||||
body_hint="todo: rotate keys",
|
||||
)
|
||||
defaults.update(kw)
|
||||
return Plan(**defaults)
|
||||
|
||||
|
||||
def test_plan_is_frozen() -> None:
|
||||
p = _plan()
|
||||
with pytest.raises(Exception): # FrozenInstanceError or AttributeError
|
||||
p.persona = "ubuntu" # type: ignore[misc]
|
||||
|
||||
|
||||
def test_edit_plan_requires_previous_body() -> None:
|
||||
with pytest.raises(ValueError, match="previous_body"):
|
||||
_plan(action="edit", previous_body=None)
|
||||
|
||||
|
||||
def test_edit_plan_with_previous_body_succeeds() -> None:
|
||||
p = _plan(action="edit", previous_body="- [ ] rotate keys\n")
|
||||
assert p.action == "edit"
|
||||
assert p.previous_body == "- [ ] rotate keys\n"
|
||||
|
||||
|
||||
def test_create_plan_does_not_need_previous_body() -> None:
|
||||
p = _plan(action="create", previous_body=None)
|
||||
assert p.action == "create"
|
||||
assert p.previous_body is None
|
||||
Reference in New Issue
Block a user