feat(projection): add complete observation day gate (issue #94)

This commit is contained in:
xavierk
2026-09-18 15:23:59 +05:30
parent 95cca2e115
commit b76067104b
6 changed files with 461 additions and 1 deletions
+49
View File
@@ -483,6 +483,31 @@ def _build_pu_context_line(conn, rate, days, clock_now):
return "Percentage Used: %d%%" % pu return "Percentage Used: %d%%" % pu
# ---------------------------------------------------------------------------
# Complete observation day gate (issue #94)
# ---------------------------------------------------------------------------
def _has_complete_local_day(conn):
"""Check if at least one complete local observation day exists.
A complete local day is a full midnight-to-midnight calendar day
within a monitoring period that has usable observation evidence.
This is the prerequisite for showing an endurance outlook.
"""
row = conn.execute(
"SELECT 1 FROM local_days WHERE complete = 1 LIMIT 1"
).fetchone()
return row is not None
def _count_complete_local_days(conn):
"""Count the number of complete local observation days."""
row = conn.execute(
"SELECT COUNT(*) FROM local_days WHERE complete = 1"
).fetchone()
return row[0] if row else 0
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Main projection function # Main projection function
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -499,6 +524,30 @@ def compute_projection(conn, clock_now):
tier, baseline, baseline_label, baseline_facts = _resolve_baseline(conn, current_segment) tier, baseline, baseline_label, baseline_facts = _resolve_baseline(conn, current_segment)
facts.extend(baseline_facts) facts.extend(baseline_facts)
# --- Complete observation day gate (issue #94) ---
# An endurance outlook requires at least one complete local
# midnight-to-midnight calendar day with usable observation evidence.
has_complete_day = _has_complete_local_day(conn)
if not has_complete_day:
facts.append("waiting for a full local observation day")
return ProjectionResult(
confidence_state=ConfidenceState.UNSUPPORTED,
contributing_facts=facts,
headline_remaining_seconds=None,
scenario_range=None,
pu_context_line="Percentage Used: unknown" if _get_latest_pu(conn) is None else "Percentage Used: %d%%" % (_get_latest_pu(conn) or 0),
disclosure_text=list(DISCLOSURES),
baseline_tier=tier,
baseline_label=baseline_label,
regime_days=None,
habit_change_fact=None,
warming_fact=None,
staleness_fact=None,
degraded_identity_fact=None,
zero_rate_fact=None,
qualifying_days_progress=None,
)
segment_days = _get_days_in_segment(conn, current_segment["opened_at"]) if current_segment else _get_all_days(conn) segment_days = _get_days_in_segment(conn, current_segment["opened_at"]) if current_segment else _get_all_days(conn)
all_days = _get_all_days(conn) all_days = _get_all_days(conn)
+37 -1
View File
@@ -109,12 +109,40 @@ def _open_period(conn, start="2026-09-01T00:00:00+00:00"):
ensure_period_open(conn, datetime.fromisoformat(start)) ensure_period_open(conn, datetime.fromisoformat(start))
def _insert_local_day(conn, local_date, tz_name="UTC", tz_offset="+00:00",
utc_start=None, utc_end=None, bw=1024*1024*100,
br=0, coverage=0.95, samples=24, complete=True):
"""Insert a local_days row (issue #94 gate prerequisite)."""
if utc_start is None:
utc_start = local_date + "T00:00:00+00:00"
if utc_end is None:
dt = datetime.strptime(local_date, "%Y-%m-%d") + timedelta(days=1)
utc_end = dt.strftime("%Y-%m-%dT00:00:00+00:00")
conn.execute(
"INSERT INTO local_days "
"(local_date, tz_name, tz_offset, utc_start, utc_end, "
" bytes_written, bytes_read, coverage, sample_count, complete) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(local_date, tz_name, tz_offset, utc_start, utc_end,
bw, br, coverage, samples, complete),
)
conn.commit()
def _insert_complete_local_days(conn, start_date, count, bw=1024*1024*100):
"""Insert multiple complete local days to satisfy the issue #94 gate."""
for i in range(count):
d = (datetime.strptime(start_date, "%Y-%m-%d") + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_local_day(conn, d, bw=bw)
def _setup_full_store(conn, *, baseline=True, segment=True, days=30, def _setup_full_store(conn, *, baseline=True, segment=True, days=30,
bw=1024*1024*100, coverage=0.95, samples_per_day=24, bw=1024*1024*100, coverage=0.95, samples_per_day=24,
sample_ts="2026-09-30T10:00:00+00:00", sample_ts="2026-09-30T10:00:00+00:00",
period_start="2026-09-01T00:00:00+00:00", period_start="2026-09-01T00:00:00+00:00",
segment_opened="2026-09-01T00:00:00+00:00", segment_opened="2026-09-01T00:00:00+00:00",
baseline_kw=None, segment_kw=None): baseline_kw=None, segment_kw=None,
local_days=True):
if baseline: if baseline:
_insert_baseline(conn, **(baseline_kw or {})) _insert_baseline(conn, **(baseline_kw or {}))
if segment: if segment:
@@ -125,6 +153,9 @@ def _setup_full_store(conn, *, baseline=True, segment=True, days=30,
_insert_day(conn, d, bw=bw, coverage=coverage, samples=samples_per_day) _insert_day(conn, d, bw=bw, coverage=coverage, samples=samples_per_day)
if sample_ts: if sample_ts:
_insert_sample(conn, sample_ts) _insert_sample(conn, sample_ts)
# Issue #94: satisfy the complete-observation-day gate
if local_days and days > 0:
_insert_complete_local_days(conn, "2026-09-29", 1, bw=bw)
# =================================================================== # ===================================================================
@@ -217,6 +248,7 @@ class TestCI1StateMatrix:
d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(conn, d, bw=1024*1024*100) _insert_day(conn, d, bw=1024*1024*100)
_insert_sample(conn, "2026-09-30T10:00:00+00:00") _insert_sample(conn, "2026-09-30T10:00:00+00:00")
_insert_complete_local_days(conn, "2026-09-29", 1)
proj = compute_projection(conn, _clock()) proj = compute_projection(conn, _clock())
assert proj.confidence_state == ConfidenceState.LIMITED assert proj.confidence_state == ConfidenceState.LIMITED
conn.close() conn.close()
@@ -230,6 +262,7 @@ class TestCI1StateMatrix:
d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24) _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24)
_insert_sample(conn, "2026-09-30T10:00:00+00:00") _insert_sample(conn, "2026-09-30T10:00:00+00:00")
_insert_complete_local_days(conn, "2026-09-29", 1)
proj = compute_projection(conn, _clock()) proj = compute_projection(conn, _clock())
assert proj.confidence_state == ConfidenceState.LIMITED assert proj.confidence_state == ConfidenceState.LIMITED
assert proj.warming_fact is not None assert proj.warming_fact is not None
@@ -245,6 +278,7 @@ class TestCI1StateMatrix:
_insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24) _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24)
stale_ts = (_clock() - timedelta(days=5)).isoformat() stale_ts = (_clock() - timedelta(days=5)).isoformat()
_insert_sample(conn, stale_ts) _insert_sample(conn, stale_ts)
_insert_complete_local_days(conn, "2026-08-30", 1)
proj = compute_projection(conn, _clock()) proj = compute_projection(conn, _clock())
assert proj.confidence_state == ConfidenceState.LIMITED assert proj.confidence_state == ConfidenceState.LIMITED
assert proj.staleness_fact is not None assert proj.staleness_fact is not None
@@ -259,6 +293,7 @@ class TestCI1StateMatrix:
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24) _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24)
_insert_sample(conn, "2026-09-30T10:00:00+00:00") _insert_sample(conn, "2026-09-30T10:00:00+00:00")
_insert_complete_local_days(conn, "2026-09-29", 1)
proj = compute_projection(conn, _clock()) proj = compute_projection(conn, _clock())
assert proj.confidence_state == ConfidenceState.LIMITED assert proj.confidence_state == ConfidenceState.LIMITED
assert proj.degraded_identity_fact is not None assert proj.degraded_identity_fact is not None
@@ -273,6 +308,7 @@ class TestCI1StateMatrix:
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(conn, d, bw=0) _insert_day(conn, d, bw=0)
_insert_sample(conn, "2026-09-30T10:00:00+00:00") _insert_sample(conn, "2026-09-30T10:00:00+00:00")
_insert_complete_local_days(conn, "2026-09-29", 1)
proj = compute_projection(conn, _clock()) proj = compute_projection(conn, _clock())
assert proj.confidence_state == ConfidenceState.UNSUPPORTED assert proj.confidence_state == ConfidenceState.UNSUPPORTED
assert proj.zero_rate_fact is not None assert proj.zero_rate_fact is not None
+258
View File
@@ -0,0 +1,258 @@
"""Complete observation day gate tests (issue #94).
Verifies that the endurance projection is withheld until at least one
complete local calendar day has been observed within a monitoring period.
Seams:
- compute_projection() → gate check via local_days table
- ProjectionResult.contributing_facts → "waiting for a full local observation day"
Acceptance criteria:
- Gate-1: No complete local day → UNSUPPORTED with waiting fact
- Gate-2: One complete local day → Limited confidence (if other conditions met)
- Gate-3: Partial days don't satisfy the gate
- Gate-4: CLI and TUI share the same gate via compute_projection()
"""
import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path
import pytest
import sys
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.store import init_store
from fenris.monitoring_periods import ensure_period_open
from fenris.projection import (
compute_projection, ConfidenceState, BaselineTier,
WARMING_COVERAGE_FLOOR,
)
@pytest.fixture
def store(tmp_path):
conn = init_store(tmp_path / "test.db")
yield conn
conn.close()
def _clock(year=2026, month=9, day=30, hour=12):
return datetime(year, month, day, hour, 0, 0, tzinfo=timezone.utc)
def _insert_baseline(conn, tbw_tb=1.0, verified=True):
conn.execute(
"INSERT INTO endurance_baseline "
"(tbw_terabytes, source_url, document_revision, entry_date, model_string, "
" nominal_capacity_bytes, validated_by, verified, created_at, updated_at) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(tbw_tb, "https://example.com/spec", "v1.0", "2026-01-01",
"Samsung SSD 970 EVO Plus 1TB", 1024000000000,
"machine_match" if verified else None, verified,
"2026-01-01T00:00:00+00:00", "2026-01-01T00:00:00+00:00"),
)
conn.commit()
def _insert_segment(conn, opened_at="2026-09-01T00:00:00+00:00"):
conn.execute(
"INSERT INTO controller_segments "
"(opened_at, identity_key, identity_degraded, subnqn, sn, mn, fr, vid, ssvid, transport) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(opened_at, "nqn.test", False, "nqn.test", "SN123",
"Samsung SSD 970 EVO Plus 1TB", "FW1", "0x144d", "0x144d", "pcie"),
)
conn.commit()
def _insert_day(conn, day, bw=1024*1024*100, coverage=0.95, samples=24):
conn.execute(
"INSERT INTO day_aggregates (day, active_seconds, idle_seconds, powered_off_seconds, "
"unknown_seconds, bytes_written_delta, bytes_read_delta, sample_count, coverage) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
(day, 3600, 0, 0, 0, bw, 0, samples, coverage),
)
conn.commit()
def _insert_sample(conn, ts, pu=5):
conn.execute(
"INSERT INTO samples (ts, device, data_units_written, data_units_read, "
"percentage_used, bytes_written, bytes_read, power_on_hours) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(ts, "/dev/nvme0n1", 1000000, 500000, pu, 512000000000, 256000000000, 8765),
)
conn.commit()
def _insert_local_day(conn, local_date, tz_name="UTC", tz_offset="+00:00",
utc_start=None, utc_end=None, bw=1024*1024*100,
br=0, coverage=0.95, samples=24, complete=True):
"""Insert a local_days row for testing the gate."""
if utc_start is None:
utc_start = local_date + "T00:00:00+00:00"
if utc_end is None:
# Next day
dt = datetime.strptime(local_date, "%Y-%m-%d") + timedelta(days=1)
utc_end = dt.strftime("%Y-%m-%dT00:00:00+00:00")
conn.execute(
"INSERT INTO local_days "
"(local_date, tz_name, tz_offset, utc_start, utc_end, "
" bytes_written, bytes_read, coverage, sample_count, complete) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(local_date, tz_name, tz_offset, utc_start, utc_end,
bw, br, coverage, samples, complete),
)
conn.commit()
def _open_period(conn, start="2026-09-01T00:00:00+00:00"):
ensure_period_open(conn, datetime.fromisoformat(start))
# ---------------------------------------------------------------------------
# Gate-1: No complete local day → UNSUPPORTED with waiting fact
# ---------------------------------------------------------------------------
class TestGateNoCompleteDay:
"""Projection is unavailable before any complete local observation day."""
def test_no_local_days_unsupported(self, store):
"""With no local_days entries, projection is UNSUPPORTED."""
_insert_baseline(store)
_insert_segment(store)
_open_period(store)
# 14 days of UTC data — enough for normal projection, but no local_days
for i in range(14):
d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=1024*1024*100)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.UNSUPPORTED
assert any("full local observation day" in f for f in result.contributing_facts)
assert result.headline_remaining_seconds is None
def test_only_partial_local_days_unsupported(self, store):
"""Partial (incomplete) local days don't satisfy the gate."""
_insert_baseline(store)
_insert_segment(store)
_open_period(store)
for i in range(14):
d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=1024*1024*100)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
# Insert only incomplete local days
for i in range(5):
d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_local_day(store, d, complete=False, coverage=0.3)
result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.UNSUPPORTED
assert any("full local observation day" in f for f in result.contributing_facts)
def test_gate_before_warming_check(self, store):
"""Gate fires even when warming would also block — gate has precedence."""
_insert_baseline(store)
_insert_segment(store)
_open_period(store)
# Only 3 days of data (below warming threshold)
for i in range(3):
d = (datetime(2026, 9, 27) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=1024*1024*100)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
# Insert one complete local day — but still below warming
_insert_local_day(store, "2026-09-29", complete=True)
result = compute_projection(store, _clock())
# Gate is satisfied (one complete day), but warming blocks Supported
# The key assertion: gate message should NOT appear when gate IS met
assert not any("full local observation day" in f for f in result.contributing_facts)
# ---------------------------------------------------------------------------
# Gate-2: One complete local day → Limited confidence
# ---------------------------------------------------------------------------
class TestGateOneCompleteDay:
"""After one complete local day, projection can proceed with Limited confidence."""
def test_one_complete_day_allows_projection(self, store):
"""With one complete local day and valid baseline/rate, projection is Limited."""
_insert_baseline(store, tbw_tb=1.0, verified=True)
_insert_segment(store)
_open_period(store)
# 14 days of UTC data
for i in range(14):
d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=1024*1024*100)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
# One complete local day
_insert_local_day(store, "2026-09-29", complete=True)
result = compute_projection(store, _clock())
# Gate satisfied — no "waiting" fact
assert not any("full local observation day" in f for f in result.contributing_facts)
# With only 14 days and other Limited factors, should be Limited or Supported
assert result.confidence_state in (ConfidenceState.LIMITED, ConfidenceState.SUPPORTED)
# Headline should exist (rate > 0, baseline exists)
assert result.headline_remaining_seconds is not None
def test_gate_fact_absent_when_satisfied(self, store):
"""The 'waiting for full day' fact does not appear when gate is met."""
_insert_baseline(store)
_insert_segment(store)
_open_period(store)
for i in range(20):
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=1024*1024*100)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_local_day(store, "2026-09-29", complete=True)
result = compute_projection(store, _clock())
assert not any("full local observation day" in f for f in result.contributing_facts)
assert result.confidence_state != ConfidenceState.UNSUPPORTED
# ---------------------------------------------------------------------------
# Gate-3: Partial first day doesn't satisfy the gate
# ---------------------------------------------------------------------------
class TestGatePartialFirstDay:
"""Starting monitoring at noon means the partial first day doesn't count."""
def test_partial_first_day_not_enough(self, store):
"""A single incomplete local day (started at noon) doesn't open the gate."""
_insert_baseline(store)
_insert_segment(store, opened_at="2026-09-29T12:00:00+00:00")
_open_period(store, start="2026-09-29T12:00:00+00:00")
# Only Sep 29 (partial) and Sep 30 (today, partial)
_insert_day(store, "2026-09-29", bw=1024*1024*100, coverage=0.5, samples=12)
_insert_day(store, "2026-09-30", bw=1024*1024*100, coverage=0.5, samples=12)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
# Only partial local days
_insert_local_day(store, "2026-09-29", complete=False, coverage=0.5)
_insert_local_day(store, "2026-09-30", complete=False, coverage=0.5)
result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.UNSUPPORTED
assert any("full local observation day" in f for f in result.contributing_facts)
# ---------------------------------------------------------------------------
# Gate-4: Multiple complete days also satisfy the gate
# ---------------------------------------------------------------------------
class TestGateMultipleCompleteDays:
"""Multiple complete local days satisfy the gate."""
def test_multiple_complete_days_satisfy_gate(self, store):
"""Several complete local days open the gate."""
_insert_baseline(store)
_insert_segment(store)
_open_period(store)
for i in range(14):
d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=1024*1024*100)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
# 7 complete local days
for i in range(7):
d = (datetime(2026, 9, 23) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_local_day(store, d, complete=True)
result = compute_projection(store, _clock())
assert not any("full local observation day" in f for f in result.contributing_facts)
assert result.headline_remaining_seconds is not None
+35
View File
@@ -86,6 +86,33 @@ def _open_period(conn, start="2026-09-01T00:00:00+00:00"):
conn.commit() conn.commit()
def _insert_local_day(conn, local_date, tz_name="UTC", tz_offset="+00:00",
utc_start=None, utc_end=None, bw=1024*1024*100,
br=0, coverage=0.95, samples=24, complete=True):
"""Insert a local_days row (issue #94 gate prerequisite)."""
if utc_start is None:
utc_start = local_date + "T00:00:00+00:00"
if utc_end is None:
dt = datetime.strptime(local_date, "%Y-%m-%d") + timedelta(days=1)
utc_end = dt.strftime("%Y-%m-%dT00:00:00+00:00")
conn.execute(
"INSERT INTO local_days "
"(local_date, tz_name, tz_offset, utc_start, utc_end, "
" bytes_written, bytes_read, coverage, sample_count, complete) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(local_date, tz_name, tz_offset, utc_start, utc_end,
bw, br, coverage, samples, complete),
)
conn.commit()
def _insert_complete_local_days(conn, start_date, count, bw=1024*1024*100):
"""Insert multiple complete local days to satisfy the issue #94 gate."""
for i in range(count):
d = (datetime.strptime(start_date, "%Y-%m-%d") + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_local_day(conn, d, bw=bw)
class TestHonestQualifyingProgress: class TestHonestQualifyingProgress:
"""Issue #77: Show honest qualifying-day progress and confidence.""" """Issue #77: Show honest qualifying-day progress and confidence."""
@@ -103,6 +130,7 @@ class TestHonestQualifyingProgress:
_insert_day(store, d, bw=bw, coverage=cov) _insert_day(store, d, bw=bw, coverage=cov)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
# After warming, qualifying_days_progress shows honest count # After warming, qualifying_days_progress shows honest count
@@ -125,6 +153,7 @@ class TestHonestQualifyingProgress:
_insert_day(store, d, bw=bw, samples=samples) _insert_day(store, d, bw=bw, samples=samples)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
# After warming, qualifying_days_progress shows honest count # After warming, qualifying_days_progress shows honest count
@@ -146,6 +175,7 @@ class TestHonestQualifyingProgress:
_insert_day(store, d, bw=bw, coverage=0.95) _insert_day(store, d, bw=bw, coverage=0.95)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
# Should be warming # Should be warming
@@ -164,6 +194,7 @@ class TestHonestQualifyingProgress:
_insert_day(store, d, bw=0, coverage=0.95) _insert_day(store, d, bw=0, coverage=0.95)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
# Zero rate should be UNSUPPORTED # Zero rate should be UNSUPPORTED
@@ -185,6 +216,7 @@ class TestHonestQualifyingProgress:
_insert_day(store, d, bw=bw, coverage=cov) _insert_day(store, d, bw=bw, coverage=cov)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
# Should still be warming (3 days below coverage > WARMING_MAX_LOW_COVERAGE=2) # Should still be warming (3 days below coverage > WARMING_MAX_LOW_COVERAGE=2)
@@ -205,6 +237,7 @@ class TestHonestQualifyingProgress:
_insert_day(store, d, bw=bw, samples=samples) _insert_day(store, d, bw=bw, samples=samples)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
# Should still be warming # Should still be warming
@@ -225,6 +258,7 @@ class TestHonestQualifyingProgress:
_insert_day(store, d, bw=bw, coverage=cov) _insert_day(store, d, bw=bw, coverage=cov)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
# Should not be warming (14 total, 2 below coverage <= WARMING_MAX_LOW_COVERAGE) # Should not be warming (14 total, 2 below coverage <= WARMING_MAX_LOW_COVERAGE)
@@ -247,6 +281,7 @@ class TestHonestQualifyingProgress:
_insert_day(store, d, bw=bw, coverage=0.95) _insert_day(store, d, bw=bw, coverage=0.95)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
# Should not be warming # Should not be warming
+53
View File
@@ -88,6 +88,33 @@ def _open_period(conn, start="2026-09-01T00:00:00+00:00"):
ensure_period_open(conn, datetime.fromisoformat(start)) ensure_period_open(conn, datetime.fromisoformat(start))
def _insert_local_day(conn, local_date, tz_name="UTC", tz_offset="+00:00",
utc_start=None, utc_end=None, bw=1024*1024*100,
br=0, coverage=0.95, samples=24, complete=True):
"""Insert a local_days row (issue #94 gate prerequisite)."""
if utc_start is None:
utc_start = local_date + "T00:00:00+00:00"
if utc_end is None:
dt = datetime.strptime(local_date, "%Y-%m-%d") + timedelta(days=1)
utc_end = dt.strftime("%Y-%m-%dT00:00:00+00:00")
conn.execute(
"INSERT INTO local_days "
"(local_date, tz_name, tz_offset, utc_start, utc_end, "
" bytes_written, bytes_read, coverage, sample_count, complete) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(local_date, tz_name, tz_offset, utc_start, utc_end,
bw, br, coverage, samples, complete),
)
conn.commit()
def _insert_complete_local_days(conn, start_date, count, bw=1024*1024*100):
"""Insert multiple complete local days to satisfy the issue #94 gate."""
for i in range(count):
d = (datetime.strptime(start_date, "%Y-%m-%d") + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_local_day(conn, d, bw=bw)
class TestPrecedence: class TestPrecedence:
def test_no_baseline_unavailable(self, store): def test_no_baseline_unavailable(self, store):
_insert_segment(store) _insert_segment(store)
@@ -127,6 +154,7 @@ class TestConfidenceRuleTable:
def test_unavailable_no_baseline(self, store): def test_unavailable_no_baseline(self, store):
_insert_segment(store) _insert_segment(store)
_open_period(store) _open_period(store)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.UNSUPPORTED assert result.confidence_state == ConfidenceState.UNSUPPORTED
assert any("no applicable endurance baseline" in f for f in result.contributing_facts) assert any("no applicable endurance baseline" in f for f in result.contributing_facts)
@@ -138,6 +166,7 @@ class TestConfidenceRuleTable:
for i in range(20): for i in range(20):
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=0) _insert_day(store, d, bw=0)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.UNSUPPORTED assert result.confidence_state == ConfidenceState.UNSUPPORTED
assert any("no finite projection" in f for f in result.contributing_facts) assert any("no finite projection" in f for f in result.contributing_facts)
@@ -150,12 +179,14 @@ class TestConfidenceRuleTable:
d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=1024*1024*100) _insert_day(store, d, bw=1024*1024*100)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.LIMITED assert result.confidence_state == ConfidenceState.LIMITED
assert any("regime only" in f and "days old" in f for f in result.contributing_facts) assert any("regime only" in f and "days old" in f for f in result.contributing_facts)
def test_limited_degraded_identity(self, store): def test_limited_degraded_identity(self, store):
_insert_baseline(store, tbw_tb=1.0, verified=True) _insert_baseline(store, tbw_tb=1.0, verified=True)
_insert_complete_local_days(store, "2026-09-29", 1)
_insert_segment(store, degraded=True) _insert_segment(store, degraded=True)
_open_period(store) _open_period(store)
for i in range(20): for i in range(20):
@@ -201,6 +232,7 @@ class TestZeroRate:
for i in range(20): for i in range(20):
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=0) _insert_day(store, d, bw=0)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.UNSUPPORTED assert result.confidence_state == ConfidenceState.UNSUPPORTED
assert any("no finite projection from this history" in f for f in result.contributing_facts) assert any("no finite projection from this history" in f for f in result.contributing_facts)
@@ -385,6 +417,7 @@ class TestSustainedRegimeRate:
d = (datetime(2026, 6, 1) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 6, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=bw) _insert_day(store, d, bw=bw)
_insert_sample(store, "2026-09-28T12:00:00+00:00", pu=5) _insert_sample(store, "2026-09-28T12:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock(year=2026, month=9, day=30, hour=12)) result = compute_projection(store, _clock(year=2026, month=9, day=30, hour=12))
# Regime should be capped at 90 days (from Jun 1 to Sep 30 = 90 days at cutoff) # Regime should be capped at 90 days (from Jun 1 to Sep 30 = 90 days at cutoff)
# The 90-day cutoff is Sep 30 - 90 = Jul 1, so regime starts Jul 1 # The 90-day cutoff is Sep 30 - 90 = Jul 1, so regime starts Jul 1
@@ -447,6 +480,7 @@ class TestHabitChange:
d = (datetime(2026, 8, 29) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 8, 29) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=bw_high) _insert_day(store, d, bw=bw_high)
_insert_sample(store, "2026-09-08T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-08T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-07", 1)
result = compute_projection(store, _clock(year=2026, month=9, day=8, hour=12)) result = compute_projection(store, _clock(year=2026, month=9, day=8, hour=12))
assert result.habit_change_fact is not None assert result.habit_change_fact is not None
assert "usage habit changed" in result.habit_change_fact assert "usage habit changed" in result.habit_change_fact
@@ -468,6 +502,7 @@ class TestHabitChange:
d = (datetime(2026, 8, 29) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 8, 29) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=bw_low) _insert_day(store, d, bw=bw_low)
_insert_sample(store, "2026-09-08T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-08T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-07", 1)
result = compute_projection(store, _clock(year=2026, month=9, day=8, hour=12)) result = compute_projection(store, _clock(year=2026, month=9, day=8, hour=12))
assert result.habit_change_fact is not None assert result.habit_change_fact is not None
assert "usage habit changed" in result.habit_change_fact assert "usage habit changed" in result.habit_change_fact
@@ -524,6 +559,7 @@ class TestHabitChange:
d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=bw) _insert_day(store, d, bw=bw)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.LIMITED assert result.confidence_state == ConfidenceState.LIMITED
assert any("regime only" in f and "days old" in f for f in result.contributing_facts) assert any("regime only" in f and "days old" in f for f in result.contributing_facts)
@@ -544,6 +580,7 @@ class TestWarmingGate:
d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=bw, coverage=0.95) _insert_day(store, d, bw=bw, coverage=0.95)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.warming_fact is not None assert result.warming_fact is not None
assert "warming up" in result.warming_fact assert "warming up" in result.warming_fact
@@ -560,6 +597,7 @@ class TestWarmingGate:
cov = 0.30 if i < 3 else 0.95 cov = 0.30 if i < 3 else 0.95
_insert_day(store, d, bw=bw, coverage=cov) _insert_day(store, d, bw=bw, coverage=cov)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.warming_fact is not None assert result.warming_fact is not None
assert "warming up" in result.warming_fact assert "warming up" in result.warming_fact
@@ -589,6 +627,7 @@ class TestWarmingGate:
cov = 0.30 if i < 3 else 0.95 cov = 0.30 if i < 3 else 0.95
_insert_day(store, d, bw=bw, coverage=cov) _insert_day(store, d, bw=bw, coverage=cov)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.warming_fact is not None assert result.warming_fact is not None
@@ -602,6 +641,7 @@ class TestWarmingGate:
d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=bw, coverage=0.95) _insert_day(store, d, bw=bw, coverage=0.95)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
# Should have warming fact but still render # Should have warming fact but still render
assert result.warming_fact is not None assert result.warming_fact is not None
@@ -614,6 +654,7 @@ class TestWarmingGate:
_insert_segment(store) _insert_segment(store)
_open_period(store) _open_period(store)
_insert_day(store, "2026-09-28", bw=100*1024*1024) _insert_day(store, "2026-09-28", bw=100*1024*1024)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.UNSUPPORTED assert result.confidence_state == ConfidenceState.UNSUPPORTED
assert result.headline_remaining_seconds is None assert result.headline_remaining_seconds is None
@@ -626,6 +667,7 @@ class TestWarmingGate:
for i in range(20): for i in range(20):
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=0) _insert_day(store, d, bw=0)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.UNSUPPORTED assert result.confidence_state == ConfidenceState.UNSUPPORTED
assert result.headline_remaining_seconds is None assert result.headline_remaining_seconds is None
@@ -646,6 +688,7 @@ class TestStalenessDrop:
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=bw, coverage=0.95) _insert_day(store, d, bw=bw, coverage=0.95)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
# Clock is 3 days after last data → staleness > 48h # Clock is 3 days after last data → staleness > 48h
clock = datetime(2026, 10, 3, 12, 0, 0, tzinfo=timezone.utc) clock = datetime(2026, 10, 3, 12, 0, 0, tzinfo=timezone.utc)
result = compute_projection(store, clock) result = compute_projection(store, clock)
@@ -661,6 +704,7 @@ class TestStalenessDrop:
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=bw, coverage=0.95) _insert_day(store, d, bw=bw, coverage=0.95)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
clock = datetime(2026, 10, 3, 12, 0, 0, tzinfo=timezone.utc) clock = datetime(2026, 10, 3, 12, 0, 0, tzinfo=timezone.utc)
result = compute_projection(store, clock) result = compute_projection(store, clock)
assert result.staleness_fact is not None assert result.staleness_fact is not None
@@ -703,6 +747,7 @@ class TestSegmentBreakProjection:
d = (datetime(2026, 9, 16) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 16) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=bw) _insert_day(store, d, bw=bw)
_insert_sample(store, "2026-09-20T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-20T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-19", 1)
result = compute_projection(store, _clock(year=2026, month=9, day=20, hour=12)) result = compute_projection(store, _clock(year=2026, month=9, day=20, hour=12))
# Prior days exist in store but projection uses current segment # Prior days exist in store but projection uses current segment
# 5 days in segment → regime_days = 5 # 5 days in segment → regime_days = 5
@@ -727,6 +772,7 @@ class TestSegmentBreakProjection:
d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=bw) _insert_day(store, d, bw=bw)
_insert_sample(store, "2026-09-28T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-28T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-27", 1)
result = compute_projection(store, _clock(year=2026, month=9, day=28, hour=12)) result = compute_projection(store, _clock(year=2026, month=9, day=28, hour=12))
# Young regime (3 days) → Limited, not enough data for full confidence # Young regime (3 days) → Limited, not enough data for full confidence
assert result.confidence_state == ConfidenceState.LIMITED assert result.confidence_state == ConfidenceState.LIMITED
@@ -753,6 +799,7 @@ class TestSegmentBreakProjection:
d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=bw) _insert_day(store, d, bw=bw)
_insert_sample(store, "2026-09-28T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-28T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-27", 1)
result = compute_projection(store, _clock(year=2026, month=9, day=28, hour=12)) result = compute_projection(store, _clock(year=2026, month=9, day=28, hour=12))
# Prior history quarantined; only 3 days in new segment # Prior history quarantined; only 3 days in new segment
assert result.regime_days is not None assert result.regime_days is not None
@@ -771,6 +818,7 @@ class TestDegradedIdentity:
_insert_segment(store, identity_key=None, degraded=True) _insert_segment(store, identity_key=None, degraded=True)
_open_period(store) _open_period(store)
_insert_day(store, "2026-09-28", bw=100*1024*1024) _insert_day(store, "2026-09-28", bw=100*1024*1024)
_insert_complete_local_days(store, "2026-09-29", 1)
# No baseline → Unavailable # No baseline → Unavailable
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.UNSUPPORTED assert result.confidence_state == ConfidenceState.UNSUPPORTED
@@ -787,6 +835,7 @@ class TestDegradedIdentity:
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=bw, coverage=0.95) _insert_day(store, d, bw=bw, coverage=0.95)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.LIMITED assert result.confidence_state == ConfidenceState.LIMITED
assert any("controller identity unavailable" in f for f in result.contributing_facts) assert any("controller identity unavailable" in f for f in result.contributing_facts)
@@ -802,6 +851,7 @@ class TestDegradedIdentity:
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=bw, coverage=0.95) _insert_day(store, d, bw=bw, coverage=0.95)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
# Stale clock (>48h) # Stale clock (>48h)
clock = datetime(2026, 10, 5, 12, 0, 0, tzinfo=timezone.utc) clock = datetime(2026, 10, 5, 12, 0, 0, tzinfo=timezone.utc)
result = compute_projection(store, clock) result = compute_projection(store, clock)
@@ -857,6 +907,7 @@ class TestIdentityChangeBlankKeys:
d = (datetime(2026, 9, 21) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 21) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=bw) _insert_day(store, d, bw=bw)
_insert_sample(store, "2026-09-26T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-26T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-25", 1)
result = compute_projection(store, _clock(year=2026, month=9, day=26, hour=12)) result = compute_projection(store, _clock(year=2026, month=9, day=26, hour=12))
# Prior history quarantined; only 5 days in new segment # Prior history quarantined; only 5 days in new segment
assert result.regime_days is not None assert result.regime_days is not None
@@ -880,6 +931,7 @@ class TestIdentityChangeBlankKeys:
d = (datetime(2026, 9, 21) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 21) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=bw) _insert_day(store, d, bw=bw)
_insert_sample(store, "2026-09-26T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-26T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-25", 1)
result = compute_projection(store, _clock(year=2026, month=9, day=26, hour=12)) result = compute_projection(store, _clock(year=2026, month=9, day=26, hour=12))
assert result.regime_days is not None assert result.regime_days is not None
assert result.regime_days <= 5 assert result.regime_days <= 5
@@ -895,6 +947,7 @@ class TestIdentityChangeBlankKeys:
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=bw) _insert_day(store, d, bw=bw)
_insert_sample(store, "2026-09-26T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-26T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-25", 1)
result = compute_projection(store, _clock(year=2026, month=9, day=26, hour=12)) result = compute_projection(store, _clock(year=2026, month=9, day=26, hour=12))
# All 25 days in same segment (equal blanks continue) # All 25 days in same segment (equal blanks continue)
assert result.regime_days is not None assert result.regime_days is not None
+29
View File
@@ -117,6 +117,33 @@ def _open_period(conn, start="2026-09-01T00:00:00+00:00"):
ensure_period_open(conn, datetime.fromisoformat(start)) ensure_period_open(conn, datetime.fromisoformat(start))
def _insert_local_day(conn, local_date, tz_name="UTC", tz_offset="+00:00",
utc_start=None, utc_end=None, bw=1024*1024*100,
br=0, coverage=0.95, samples=24, complete=True):
"""Insert a local_days row (issue #94 gate prerequisite)."""
if utc_start is None:
utc_start = local_date + "T00:00:00+00:00"
if utc_end is None:
dt = datetime.strptime(local_date, "%Y-%m-%d") + timedelta(days=1)
utc_end = dt.strftime("%Y-%m-%dT00:00:00+00:00")
conn.execute(
"INSERT INTO local_days "
"(local_date, tz_name, tz_offset, utc_start, utc_end, "
" bytes_written, bytes_read, coverage, sample_count, complete) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(local_date, tz_name, tz_offset, utc_start, utc_end,
bw, br, coverage, samples, complete),
)
conn.commit()
def _insert_complete_local_days(conn, start_date, count, bw=1024*1024*100):
"""Insert multiple complete local days to satisfy the issue #94 gate."""
for i in range(count):
d = (datetime.strptime(start_date, "%Y-%m-%d") + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_local_day(conn, d, bw=bw)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Unit tests for helpers # Unit tests for helpers
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -184,6 +211,7 @@ class TestStateMatrix:
d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(conn, d, bw=1024*1024*100) _insert_day(conn, d, bw=1024*1024*100)
_insert_sample(conn, "2026-09-30T10:00:00+00:00") _insert_sample(conn, "2026-09-30T10:00:00+00:00")
_insert_complete_local_days(conn, "2026-09-29", 1)
proj = compute_projection(conn, _clock()) proj = compute_projection(conn, _clock())
assert proj.confidence_state == ConfidenceState.LIMITED assert proj.confidence_state == ConfidenceState.LIMITED
assert proj.headline_remaining_seconds is not None assert proj.headline_remaining_seconds is not None
@@ -199,6 +227,7 @@ class TestStateMatrix:
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24) _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24)
_insert_sample(conn, "2026-09-30T10:00:00+00:00") _insert_sample(conn, "2026-09-30T10:00:00+00:00")
_insert_complete_local_days(conn, "2026-09-29", 1)
proj = compute_projection(conn, _clock()) proj = compute_projection(conn, _clock())
assert proj.confidence_state == ConfidenceState.SUPPORTED assert proj.confidence_state == ConfidenceState.SUPPORTED
assert proj.headline_remaining_seconds is not None assert proj.headline_remaining_seconds is not None