Sam Rolfe
1275b1baaa
Adds windows/, the Python client for the kids' Windows computers, and proves both halves of the Phase 4 definition of done on the real box (DESKTOP-TE2RHV2). The client: - subscribes, retained, to its own config on the 8883 TLS device plane with its own broker account (FR-3, FR-6); - keeps the last valid config on disk and loads it before connecting, so a broker outage changes nothing (FR-14); - terminates every process on the blocklist while the device is effectively locked, with the allowlist winning and the client's own tree plus critical system processes protected (FR-4, FR-16); - publishes presence and daily per-app usage; the day's totals are held on disk and republished when the link returns (FR-7); - shows an always-on-top warning window naming what is blocked, why and until when (FR-9). install-scheduled-task.ps1 registers the logon Scheduled Task for the kid's account with an interactive token, and grants that account Modify on the state directory: a file created from an elevated shell inherits Administrators only, and the limited task is otherwise refused with Access denied. Evidence (windows/EVIDENCE.md, from the machine's own log): notepad.exe killed inside a scheduled window (pid 16600, then 9724), the device online in the Control Service with locked/until/reason, a telemetry row for the device (app=690s), enforcement continuing with the broker unreachable (192.168.20.13:8884), the warning window on the interactive desktop, and the Scheduled Task registered at logon as New Comp. Domains are deliberately not reported: Windows has no cheap per-process DNS attribution, so the client sends "domains": [] and the gap is documented rather than guessed. 54 unit tests, run with `make windows-test` (no broker, no Windows, no psutil). AGENTS.md §11 is now 10 of 10.
116 lines
4.5 KiB
Python
116 lines
4.5 KiB
Python
"""The usage roll-up: payload shape, accumulation, pruning and sampling."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
from webrestrict.telemetry import (
|
|
MAX_DAYS,
|
|
MAX_SAMPLE_SECONDS,
|
|
TelemetryStore,
|
|
UsageSampler,
|
|
build_payload,
|
|
day_key,
|
|
to_entries,
|
|
)
|
|
|
|
NOW = datetime(2026, 10, 14, 20, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
class PayloadTests(unittest.TestCase):
|
|
def test_shape_matches_the_control_service(self):
|
|
body = build_payload("2026-10-14", {"notepad.exe": 120, "steam.exe": 60})
|
|
self.assertEqual(body["day"], "2026-10-14")
|
|
self.assertEqual(body["apps"], [{"name": "notepad.exe", "seconds": 120}, {"name": "steam.exe", "seconds": 60}])
|
|
# Windows cannot attribute domains cheaply; the gap is explicit, not guessed.
|
|
self.assertEqual(body["domains"], [])
|
|
|
|
def test_zero_and_blank_entries_are_dropped(self):
|
|
self.assertEqual(to_entries({"": 10, "steam.exe": 0, "notepad.exe": 5}), [{"name": "notepad.exe", "seconds": 5}])
|
|
|
|
def test_day_key_is_local(self):
|
|
self.assertEqual(day_key(datetime(2026, 1, 2, 23, 59)), "2026-01-02")
|
|
|
|
def test_an_aware_timestamp_is_converted_to_local_time(self):
|
|
# The day boundary must be the child's midnight, not UTC's.
|
|
offset = datetime.now().astimezone().utcoffset()
|
|
utc = datetime(2026, 1, 2, 23, 30, tzinfo=timezone.utc)
|
|
self.assertEqual(day_key(utc), (utc + offset).strftime("%Y-%m-%d"))
|
|
|
|
|
|
class StoreTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.path = Path(self.tmp.name) / "telemetry.json"
|
|
self.addCleanup(self.tmp.cleanup)
|
|
|
|
def test_accumulates_and_persists(self):
|
|
store = TelemetryStore(self.path)
|
|
store.add_seconds("2026-10-14", {"notepad.exe": 60})
|
|
store.add_seconds("2026-10-14", {"notepad.exe": 30, "steam.exe": 10})
|
|
self.assertEqual(store.day("2026-10-14"), {"notepad.exe": 90, "steam.exe": 10})
|
|
|
|
reloaded = TelemetryStore(self.path)
|
|
self.assertEqual(reloaded.day("2026-10-14"), {"notepad.exe": 90, "steam.exe": 10})
|
|
|
|
def test_ignores_non_positive_samples(self):
|
|
store = TelemetryStore(self.path)
|
|
store.add_seconds("2026-10-14", {"notepad.exe": 0, "steam.exe": -5})
|
|
self.assertEqual(store.day("2026-10-14"), {})
|
|
|
|
def test_prunes_to_the_most_recent_days(self):
|
|
store = TelemetryStore(self.path)
|
|
for day in range(1, MAX_DAYS + 5):
|
|
store.add_seconds(f"2026-01-{day:02d}", {"notepad.exe": 1})
|
|
self.assertEqual(len(store.days), MAX_DAYS)
|
|
self.assertIn("2026-01-18", store.days)
|
|
self.assertNotIn("2026-01-01", store.days)
|
|
|
|
def test_an_unreadable_file_starts_empty(self):
|
|
self.path.write_text("{not json", encoding="utf-8")
|
|
store = TelemetryStore(self.path).load()
|
|
self.assertEqual(store.days, {})
|
|
|
|
def test_written_json_is_readable(self):
|
|
store = TelemetryStore(self.path)
|
|
store.add_seconds("2026-10-14", {"notepad.exe": 5})
|
|
self.assertIn("days", json.loads(self.path.read_text(encoding="utf-8")))
|
|
|
|
|
|
class SamplerTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.store = TelemetryStore(Path(self.tmp.name) / "telemetry.json")
|
|
self.addCleanup(self.tmp.cleanup)
|
|
|
|
def test_the_first_sample_only_starts_the_clock(self):
|
|
sampler = UsageSampler(self.store)
|
|
self.assertEqual(sampler.sample(["notepad.exe"], now=NOW, day="2026-10-14"), {})
|
|
|
|
def test_credits_elapsed_seconds_to_each_visible_app(self):
|
|
sampler = UsageSampler(self.store)
|
|
sampler.sample(["notepad.exe"], now=NOW, day="2026-10-14")
|
|
totals = sampler.sample(["notepad.exe", "steam.exe"], now=NOW + timedelta(seconds=60), day="2026-10-14")
|
|
self.assertEqual(totals, {"notepad.exe": 60, "steam.exe": 60})
|
|
|
|
def test_a_long_gap_is_capped(self):
|
|
sampler = UsageSampler(self.store)
|
|
sampler.sample(["notepad.exe"], now=NOW, day="2026-10-14")
|
|
totals = sampler.sample(
|
|
["notepad.exe"], now=NOW + timedelta(hours=5), day="2026-10-14"
|
|
)
|
|
self.assertEqual(totals["notepad.exe"], int(MAX_SAMPLE_SECONDS))
|
|
|
|
def test_no_visible_apps_adds_nothing(self):
|
|
sampler = UsageSampler(self.store)
|
|
sampler.sample([], now=NOW, day="2026-10-14")
|
|
self.assertEqual(sampler.sample([], now=NOW + timedelta(seconds=60), day="2026-10-14"), {})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|