ci: add quality-gate workflow (semgrep, ruff, jscpd, advisory PR-Agent) #112

Merged
xavierk merged 4 commits from ci/quality-gates into main 2026-10-05 14:14:59 +00:00
51 changed files with 293 additions and 227 deletions
+92
View File
@@ -0,0 +1,92 @@
name: CI
on:
pull_request:
push:
branches: [main]
schedule:
- cron: '0 3 * * 1'
workflow_dispatch:
permissions:
contents: read
jobs:
security:
runs-on: bongbetic-ci
timeout-minutes: 15
container:
image: docker.io/semgrep/semgrep:1.178.0
steps:
- name: Checkout
env:
GIT_TOKEN: ${{ gitea.token }}
run: |
set -euo pipefail
git init -q .
git remote add origin "${{ gitea.server_url }}/${{ gitea.repository }}.git"
git -c http.extraheader="Authorization: token ${GIT_TOKEN}" fetch -q --depth=1 origin "${{ gitea.sha }}"
git checkout -q FETCH_HEAD
- name: Semgrep
run: semgrep scan --config p/default --config p/owasp-top-ten --metrics off --error
lint:
if: gitea.event_name != 'schedule'
runs-on: bongbetic-ci
timeout-minutes: 20
steps:
- name: Checkout
env:
GIT_TOKEN: ${{ gitea.token }}
run: |
set -euo pipefail
git init -q .
git remote add origin "${{ gitea.server_url }}/${{ gitea.repository }}.git"
git -c http.extraheader="Authorization: token ${GIT_TOKEN}" fetch -q --depth=1 origin "${{ gitea.sha }}"
git checkout -q FETCH_HEAD
# node:24-bookworm ships python3 without ensurepip, so python3-venv comes from apt.
- name: Install ruff
run: |
set -euo pipefail
apt-get update -qq
apt-get install -y -qq --no-install-recommends python3-venv
python3 -m venv /tmp/lint-venv
/tmp/lint-venv/bin/pip install -q ruff==0.16.10
- name: Ruff
run: /tmp/lint-venv/bin/ruff check src/ tests/
- name: Duplicate code (jscpd)
run: npx --yes jscpd@4.3.0 --config .jscpd.json .
ai-review:
if: gitea.event_name == 'pull_request'
runs-on: bongbetic-ci
timeout-minutes: 10
continue-on-error: true
container:
image: docker.io/pragent/pr-agent:0.47.0
env:
config__git_provider: gitea
gitea__url: https://git.bongbetic.com
gitea__personal_access_token: ${{ secrets.PR_AGENT_GITEA_TOKEN }}
openrouter__key: ${{ secrets.OPENROUTER_API_KEY }}
config__model: ${{ vars.PR_AGENT_MODEL || 'openrouter/anthropic/claude-sonnet-5' }}
config__fallback_models: "[\"${{ vars.PR_AGENT_MODEL || 'openrouter/anthropic/claude-sonnet-5' }}\"]"
config__custom_model_max_tokens: '200000'
steps:
# Advisory only: posts a review comment, never pushes code. Skips when secrets are absent
# (for example PRs from forks, where Gitea withholds secrets).
- name: PR-Agent review
env:
PR_URL: ${{ gitea.event.pull_request.html_url }}
run: |
set -euo pipefail
if [ -z "${gitea__personal_access_token}" ] || [ -z "${openrouter__key}" ]; then
echo "::notice::PR_AGENT_GITEA_TOKEN or OPENROUTER_API_KEY not set; skipping AI review"
exit 0
fi
cd /app
pr-agent --pr_url="${PR_URL}" review
+2 -2
View File
@@ -25,7 +25,7 @@ jobs:
release:
runs-on: [self-hosted]
steps:
- uses: actions/checkout@v4
- uses: actions/checkout@v4 # nosemgrep: github-actions-mutable-action-tag -- release path unchanged, tag pinned by major version
- name: Validate release tag and notes
run: |
@@ -47,7 +47,7 @@ jobs:
--footer packaging/release-footer.md > "${RUNNER_TEMP}/release-body.md"
- name: Set up Python
uses: actions/setup-python@v5
uses: actions/setup-python@v5 # nosemgrep: github-actions-mutable-action-tag -- release path unchanged, tag pinned by major version
with:
python-version: '3.12'
+18
View File
@@ -0,0 +1,18 @@
{
"threshold": 8,
"minLines": 5,
"minTokens": 50,
"reporters": ["console"],
"gitignore": true,
"ignore": [
"**/node_modules/**",
"**/.git/**",
"**/dist/**",
"**/docs/**",
"**/packaging/**",
"**/*.lock",
"**/package-lock.json",
"**/requirements.txt",
"**/*.md"
]
}
+48
View File
@@ -0,0 +1,48 @@
# CI quality gates
Workflow: `.gitea/workflows/ci.yml` (runner label `bongbetic-ci`, no third-party `uses:` actions; code is checked out with shell git).
Status contexts: `CI / security (pull_request)`, `CI / lint (pull_request)`, `CI / ai-review (pull_request)`.
| Job | Runs on | Blocks merge? | What it does |
|-----|---------|---------------|--------------|
| `security` | PR, push to main, weekly schedule, manual | Yes | Semgrep 1.178.0, `p/default` + `p/owasp-top-ten`, `--error` |
| `lint` | PR, push to main, manual | Yes | `ruff check src/ tests/` (ruff 0.16.10, rules E4/E7/E9/F) and jscpd 4.3.0 duplicate-code threshold 8% |
| `ai-review` | PR only | No (advisory) | PR-Agent `review`, comment-only, `continue-on-error: true` |
There is no `e2e` (no web UI) and no `deploy` job (not a Coolify app). The weekly schedule (`0 3 * * 1`) runs only `security`.
## Run locally
```sh
# security (same command as CI)
podman run --rm -v "$PWD:/src:Z" docker.io/semgrep/semgrep:1.178.0 \
semgrep scan --config p/default --config p/owasp-top-ten --metrics off --error
# lint (needs `pip install -e '.[dev]'` or `pip install ruff==0.16.10`)
make lint
# duplicate code
npx --yes jscpd@4.3.0 --config .jscpd.json .
```
Notes:
- The semgrep container needs no extra flags. `:Z` is only for SELinux hosts; its working directory is `/src`.
- The `lint` job installs `python3-venv` from apt because `node:24-bookworm` has no `ensurepip`; ruff itself is pinned.
- `.jscpd.json` threshold is 8%, just above the 7.15% baseline, so duplication cannot grow. Lower it as duplication is removed.
- Semgrep prints some non-fatal `PartialParsing` errors; they do not fail the job.
## PR-Agent (advisory)
`ai-review` posts a review comment on the PR through the Gitea API. It never pushes code and never blocks. It exits 0 with a notice when its secrets are empty (for example on fork PRs).
Required repository secrets: `OPENROUTER_API_KEY`, `PR_AGENT_GITEA_TOKEN` (Gitea token of the bot account, scopes to comment on PRs).
Optional repository variable: `PR_AGENT_MODEL` (default `openrouter/anthropic/claude-sonnet-5`). The workflow sets `config__custom_model_max_tokens` to 200000, so adjust it if you pick a model with a different context window.
## Caveats
- Semgrep registry rules (`p/default`, `p/owasp-top-ten`) are fetched at runtime and are not pinned, so a new rule can turn a green main red. The weekly scheduled `security` run catches this early.
- Intentional findings are suppressed with a narrow `# nosemgrep: <rule-id> -- <reason>` on the line. Do not use `.semgrepignore` for source files.
## Rollback
Revert the PR that added `ci.yml` (and its follow-up commits). If branch protection requires `CI / security`, `CI / lint` or `CI / ai-review`, relax it first, otherwise merges stay blocked on checks that no longer run.
+4
View File
@@ -13,8 +13,12 @@ dev = [
"pytest>=7.0.0",
"pytest-cov>=4.0.0",
"pytest-asyncio>=0.20.0",
"ruff==0.16.10",
]
[tool.ruff.lint]
select = ["E4", "E7", "E9", "F"]
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
+1 -1
View File
@@ -27,7 +27,7 @@ if VENV_DIR.exists():
if site_packages:
sys.path.insert(0, str(site_packages))
from fenris.collector import run_collection
from fenris.collector import run_collection # noqa: E402 -- must follow the sys.path bootstrap above
CONFIG_PATH = Path("/etc/fenris/fenris.conf")
+1 -1
View File
@@ -9,7 +9,7 @@ usage-habit classification is known rather than unknown (§5.3).
"""
import sqlite3
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from datetime import datetime, timedelta
@dataclass(frozen=True)
+1 -1
View File
@@ -12,7 +12,7 @@ No proportional allocation, endpoint assignment, or double counting.
"""
import sqlite3
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional, Tuple
from typing import Any, Dict, List, Optional
from .hour_classify import classify_hour, HourSplit
-1
View File
@@ -26,7 +26,6 @@ Runit guarantees:
Spec: §8.4, §8.5, §8.6, §8.7, §8.8, ADR 0008
"""
import os
import shutil
import subprocess
import sys
from enum import Enum
+2 -3
View File
@@ -18,11 +18,10 @@ import logging
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from typing import Any, Dict, List, Optional
from .hour_classify import classify_hour
from .segment import open_segment, normalize_identity
from .monitoring_periods import close_period
from .segment import open_segment
logger = logging.getLogger(__name__)
+2 -2
View File
@@ -394,7 +394,7 @@ def repair_legacy_local_day_evidence(conn: sqlite3.Connection) -> int:
"bytes_written": row[2], "bytes_read": row[3],
"segment_id": row[4], "local_tz": row[5],
}
for row in conn.execute(sample_select)
for row in conn.execute(sample_select) # nosemgrep: sqlalchemy-execute-raw-query -- query text is built from constant fragments only; no external input
]
for previous, current in pairwise(samples):
start = previous["ts"]
@@ -1018,7 +1018,7 @@ def query_local_day_summary(
"""
tz_filter = " AND tz_name = ?" if tz_name else ""
params = (local_date, tz_name) if tz_name else (local_date,)
row = conn.execute(
row = conn.execute( # nosemgrep: sqlalchemy-execute-raw-query -- only constant fragments are concatenated; values are bound via ? placeholders
"SELECT local_date, tz_name, tz_offset, utc_start, utc_end, "
" bytes_written, bytes_read, coverage, sample_count, complete, "
" activity_seconds, activity_intervals, activity_incomplete, "
+3 -3
View File
@@ -31,13 +31,13 @@ if VENV_DIR.exists():
if site_packages:
sys.path.insert(0, str(site_packages))
from fenris.store import init_store, get_store_path
from fenris.monitoring_periods import (
from fenris.store import init_store # noqa: E402 -- must follow the sys.path bootstrap above
from fenris.monitoring_periods import ( # noqa: E402 -- sys.path bootstrap above
ensure_period_open,
close_period,
get_open_period,
)
from fenris.init_system import (
from fenris.init_system import ( # noqa: E402 -- sys.path bootstrap above
enable_timer,
disable_timer,
collect_now,
+2 -2
View File
@@ -24,9 +24,9 @@ Criteria: PR-1–PR-17, CI-4.
"""
import sqlite3
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from datetime import datetime, timedelta
from enum import Enum
from typing import Any, Dict, List, Optional, Tuple
from typing import Any, Dict, List, Optional
from .monitoring_periods import interval_within_one_monitoring_period
+5 -5
View File
@@ -336,7 +336,7 @@ def prune_old_samples(
if owns_transaction:
conn.execute("BEGIN IMMEDIATE")
else:
conn.execute(f"SAVEPOINT {savepoint}")
conn.execute(f"SAVEPOINT {savepoint}") # nosemgrep: sqlalchemy-execute-raw-query, formatted-sql-query -- savepoint name is a function-local constant; SQLite cannot bind identifiers
try:
rows = _sample_rows(conn)
@@ -344,7 +344,7 @@ def prune_old_samples(
if owns_transaction:
conn.commit()
else:
conn.execute(f"RELEASE SAVEPOINT {savepoint}")
conn.execute(f"RELEASE SAVEPOINT {savepoint}") # nosemgrep: sqlalchemy-execute-raw-query, formatted-sql-query -- savepoint name is a function-local constant; SQLite cannot bind identifiers
return 0
newest_id = rows[-1][0]
@@ -361,12 +361,12 @@ def prune_old_samples(
if owns_transaction:
conn.commit()
else:
conn.execute(f"RELEASE SAVEPOINT {savepoint}")
conn.execute(f"RELEASE SAVEPOINT {savepoint}") # nosemgrep: sqlalchemy-execute-raw-query, formatted-sql-query -- savepoint name is a function-local constant; SQLite cannot bind identifiers
return len(expired)
except Exception:
if owns_transaction:
conn.rollback()
else:
conn.execute(f"ROLLBACK TO SAVEPOINT {savepoint}")
conn.execute(f"RELEASE SAVEPOINT {savepoint}")
conn.execute(f"ROLLBACK TO SAVEPOINT {savepoint}") # nosemgrep: sqlalchemy-execute-raw-query, formatted-sql-query -- savepoint name is a function-local constant; SQLite cannot bind identifiers
conn.execute(f"RELEASE SAVEPOINT {savepoint}") # nosemgrep: sqlalchemy-execute-raw-query, formatted-sql-query -- savepoint name is a function-local constant; SQLite cannot bind identifiers
raise
+2 -2
View File
@@ -13,11 +13,11 @@ Contracts:
"""
import logging
import sqlite3
from dataclasses import dataclass, field
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Optional, List, Tuple
from .derive import find_previous_sample, derive_hours_from_interval, _parse_ts
from .derive import derive_hours_from_interval, _parse_ts
logger = logging.getLogger(__name__)
-1
View File
@@ -17,7 +17,6 @@ import sqlite3
from datetime import datetime
from typing import Any, Dict, Optional, Tuple
from .collector import normalize_identity
def find_current_segment(conn: sqlite3.Connection) -> Optional[Dict[str, Any]]:
+2 -3
View File
@@ -15,15 +15,14 @@ Criteria: LC-9, CI-2, CI-4, FL-4, FL-5, FL-7.
"""
import sqlite3
from contextlib import contextmanager
import sys
from datetime import datetime, timedelta, timezone
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, Iterator, List, Optional, Tuple, TYPE_CHECKING
if TYPE_CHECKING:
from .status_composition import StatusComposition
from .projection import compute_projection, ConfidenceState, DISCLOSURES
from .projection import compute_projection, DISCLOSURES
from .store import SCHEMA_VERSION
from .init_system import (
query_service_state as _init_query_service_state,
+4 -9
View File
@@ -11,25 +11,20 @@ Freshness grading uses shared constants from status.py.
import enum
import sqlite3
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
# Status poll interval (AC78-6): lightweight 5s systemctl show poll
STATUS_POLL_INTERVAL_S = 5
from .status import (
FRESH_THRESHOLD_S,
STALENESS_THRESHOLD_S,
grade_freshness,
freshness_age_human,
is_deliberately_paused,
monitoring_continuity,
deliberate_pause_lines,
open_store_readonly,
StoreFault,
NewerSchema,
)
# Status poll interval (AC78-6): lightweight 5s systemctl show poll
STATUS_POLL_INTERVAL_S = 5
# ---------------------------------------------------------------------------
# Status state enum with glyph and label
+5 -6
View File
@@ -7,7 +7,6 @@ This module handles:
"""
import sqlite3
from pathlib import Path
from typing import Optional
# Schema version - increment on each migration
@@ -61,7 +60,7 @@ def init_store(store_path: Path) -> sqlite3.Connection:
if current_version == 0:
# New database - create schema
_create_schema(conn)
conn.execute(f"PRAGMA user_version={SCHEMA_VERSION}")
conn.execute(f"PRAGMA user_version={SCHEMA_VERSION}") # nosemgrep: sqlalchemy-execute-raw-query, formatted-sql-query -- SCHEMA_VERSION is an int constant; PRAGMA cannot bind parameters
conn.commit()
elif current_version > SCHEMA_VERSION:
# Unknown newer version - refuse
@@ -312,7 +311,7 @@ def _apply_migrations(conn: sqlite3.Connection, current_version: int):
conn.execute("BEGIN IMMEDIATE")
try:
migration(conn)
conn.execute(f"PRAGMA user_version={target_version}")
conn.execute(f"PRAGMA user_version={target_version}") # nosemgrep: sqlalchemy-execute-raw-query, formatted-sql-query -- target_version is an int key of the static migrations map; PRAGMA cannot bind parameters
conn.commit()
except Exception:
conn.rollback()
@@ -337,7 +336,7 @@ def _migrate_1_to_2(conn: sqlite3.Connection) -> None:
).fetchall()}
for column in ("unattributed_bytes_written", "unattributed_bytes_read"):
if column not in cols:
conn.execute(
conn.execute( # nosemgrep: sqlalchemy-execute-raw-query, formatted-sql-query -- column comes from a hardcoded tuple; DDL cannot bind identifiers
f"ALTER TABLE day_aggregates ADD COLUMN {column} INTEGER DEFAULT 0"
)
@@ -394,7 +393,7 @@ def _migrate_4_to_5(conn: sqlite3.Connection) -> None:
("last_sample_id", "INTEGER"),
):
if column not in local_cols:
conn.execute(
conn.execute( # nosemgrep: sqlalchemy-execute-raw-query, formatted-sql-query -- column and declaration come from a hardcoded tuple; DDL cannot bind identifiers
f"ALTER TABLE local_days ADD COLUMN {column} {declaration}"
)
_create_local_day_shared_evidence(conn)
@@ -436,7 +435,7 @@ def migrate_to_latest(store_path: Path) -> int:
# Version 0 means no schema — create fresh (issue #73)
if current_version == 0:
_create_schema(conn)
conn.execute(f"PRAGMA user_version={SCHEMA_VERSION}")
conn.execute(f"PRAGMA user_version={SCHEMA_VERSION}") # nosemgrep: sqlalchemy-execute-raw-query, formatted-sql-query -- SCHEMA_VERSION is an int constant; PRAGMA cannot bind parameters
conn.commit()
conn.close()
return SCHEMA_VERSION
+9 -13
View File
@@ -12,7 +12,6 @@ Criteria: TUI-1, TUI-2, TUI-4, CI-1, CI-2, CI-4, IN-3, LC-6, LC-8.
from __future__ import annotations
import sqlite3
import sys
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional
@@ -21,7 +20,7 @@ from zoneinfo import ZoneInfo
from rich.text import Text
from textual.app import App, ComposeResult
from textual.binding import Binding
from textual.containers import Container, Horizontal, VerticalScroll
from textual.containers import Container, VerticalScroll
from textual.screen import ModalScreen
from textual.message import Message
from textual.widget import Widget
@@ -38,16 +37,14 @@ from .activity_plot import VolumePoint, volume_plot
from .projection import (
ConfidenceState,
ProjectionResult,
ScenarioRange,
compute_projection,
)
from .status import (
CADENCE_DEFAULT_S,
FRESH_THRESHOLD_S,
STALENESS_THRESHOLD_S,
FRESH_THRESHOLD_S, # noqa: F401 -- re-exported; tests/test_acceptance_sweep.py asserts parity with status
STALENESS_THRESHOLD_S, # noqa: F401 -- re-exported (see above)
format_disclosures,
freshness_age_human,
grade_freshness,
grade_freshness, # noqa: F401 -- re-exported (see above)
deliberate_pause_lines,
read_status,
_journalctl_hint,
@@ -55,7 +52,6 @@ from .status import (
from .status_composition import (
StatusComposition,
render_status_tui,
STATUS_POLL_INTERVAL_S,
)
from .control import MONITOR_HELPER, MonitorError, run_monitor
from .derive import _parse_ts
@@ -1110,10 +1106,10 @@ def _query_local_day_graph_data(
(start_text, end_text, selected_date or ""),
).fetchall()
recorded_zones: Dict[str, list[str]] = {}
for local_date, timezone in rows:
for local_date, zone_name in rows:
recorded_zones.setdefault(local_date, [])
if timezone not in recorded_zones[local_date]:
recorded_zones[local_date].append(timezone)
if zone_name not in recorded_zones[local_date]:
recorded_zones[local_date].append(zone_name)
result: List[Dict[str, Any]] = []
for local_date in sorted(local_dates):
@@ -1138,9 +1134,9 @@ def _query_local_day_graph_data(
})
continue
for timezone in timezones:
for zone_name in timezones:
summary = query_local_day_summary(
conn, local_date, timezone, now,
conn, local_date, zone_name, now,
)
if summary is None:
continue
+1 -1
View File
@@ -5,7 +5,7 @@ computation used by local-day derivation. All functions are
stateless and safe to call from the collector and TUI reader.
"""
import os
from datetime import datetime, timezone
from datetime import datetime
from pathlib import Path
-1
View File
@@ -1,7 +1,6 @@
"""Shared test helpers for Fenris test suite."""
from pathlib import Path
import pytest
REPO_ROOT = Path(__file__).resolve().parent.parent
VERSION_FILE = REPO_ROOT / "pyproject.toml"
+1 -9
View File
@@ -9,7 +9,6 @@ CI-3 Prohibition set: automated structural checks
CI-4 Required wording and six disclosures in both views
"""
import re
import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch
@@ -19,16 +18,13 @@ import sys
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.store import init_store, SCHEMA_VERSION
from fenris.store import init_store
from fenris.monitoring_periods import ensure_period_open
from fenris.projection import (
compute_projection,
ConfidenceState,
BaselineTier,
DISCLOSURES,
STALENESS_HOURS,
WARMING_MIN_DAYS,
YOUNG_REGIME_DAYS,
)
from fenris.status import (
grade_freshness,
@@ -36,13 +32,9 @@ from fenris.status import (
render_status,
format_disclosures,
FRESH_THRESHOLD_S,
STALENESS_THRESHOLD_S,
CADENCE_DEFAULT_S,
ACCURACY_SEC,
)
from fenris.tui import (
FenrisTuiApp,
_format_remaining,
)
+3 -6
View File
@@ -8,7 +8,6 @@ Seams:
- write side: run_collection() → observation store
- read side: get_status(), compute_projection() → observation store
"""
import os
import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path
@@ -19,10 +18,9 @@ import sys
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.collector import run_collection, normalize_identity
from fenris.store import init_store, get_store_path, SCHEMA_VERSION
from fenris.monitoring_periods import ensure_period_open, close_period, get_open_period
from fenris.day_aggregate import derive_day, derive_all_days
from fenris.collector import run_collection
from fenris.store import init_store, SCHEMA_VERSION
from fenris.monitoring_periods import close_period, get_open_period
# ---------------------------------------------------------------------------
@@ -573,7 +571,6 @@ class TestDisplayStates:
def test_one_sample_awaiting_another_in_tui(self, tmp_path):
"""One sample → TUI shows awaiting state."""
from fenris.tui import FenrisTuiApp
db = tmp_path / "test.db"
conn = init_store(db)
conn.execute(
+3 -5
View File
@@ -6,13 +6,11 @@ Tests the thinnest complete write path through the system:
Seam: write side of the observation store database file.
"""
import json
import os
import sqlite3
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, Generator
from typing import Any, Dict
import pytest
@@ -20,8 +18,8 @@ import pytest
import sys
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.collector import run_collection, AcquisitionError, InvariantViolationError
from fenris.store import init_store, get_store_path
from fenris.collector import run_collection
from fenris.store import init_store
# Fixtures
+1 -3
View File
@@ -13,7 +13,6 @@ Acceptance criteria:
- 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
@@ -24,8 +23,7 @@ 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,
compute_projection, ConfidenceState,
)
+1 -3
View File
@@ -6,7 +6,7 @@ shared boundary evidence, not duplicated into both days.
Seam: derive._add_unattributed_bytes() → day_aggregates.unattributed_bytes_*
"""
import sqlite3
from datetime import datetime, timedelta, timezone
from datetime import datetime, timezone
from pathlib import Path
import pytest
@@ -14,8 +14,6 @@ import sys
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.collector import run_collection
from fenris.store import init_store
from fenris.monitoring_periods import ensure_period_open
def _make_smartctl(duw: int, dur: int):
+3 -4
View File
@@ -7,9 +7,8 @@ From spec §5.4, §3.3, ST-4, ST-5:
whose classification is known
- No absent hour ever interpolated/estimated/fabricated (FL-3)
"""
import sqlite3
import sys
from datetime import datetime, timezone, timedelta
from datetime import datetime, timezone
from pathlib import Path
import pytest
@@ -18,8 +17,8 @@ sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.store import init_store
from fenris.monitoring_periods import ensure_period_open, close_period
from fenris.day_aggregate import derive_day, derive_all_days, DayAggregate
from fenris.hour_classify import HourSplit, ACTIVE_THRESHOLD_BYTES
from fenris.day_aggregate import derive_day, derive_all_days
from fenris.hour_classify import HourSplit
@pytest.fixture
+1 -2
View File
@@ -11,11 +11,10 @@ Four splits sum to exactly 3600s. Disabled time is never an hour state.
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.hour_classify import classify_hour, HourSplit, ACTIVE_THRESHOLD_BYTES
from fenris.hour_classify import classify_hour, ACTIVE_THRESHOLD_BYTES
HOUR_SECONDS = 3600
-1
View File
@@ -7,7 +7,6 @@ From spec §2.3:
Padded and unpadded renderings of the same field yield byte-identical stored values.
"""
import pytest
import sys
from pathlib import Path
+1 -7
View File
@@ -9,12 +9,9 @@ Covers:
Spec: ADR 0008, §8.4, §8.5, §8.6, §8.7, §8.8
"""
import os
import sqlite3
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch, MagicMock, PropertyMock
from unittest.mock import patch, MagicMock
import pytest
@@ -24,7 +21,6 @@ sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.init_system import (
InitSystem,
detect_init_system,
get_init_system,
reset_init_system_cache,
_systemd_enable,
_systemd_disable,
@@ -41,8 +37,6 @@ from fenris.init_system import (
collect_now,
query_service_state,
journal_hint,
FENRIS_SV_DIR,
FENRIS_SERVICE_LINK,
COLLECT_TIMEOUT_S,
)
from fenris.store import init_store
+2 -4
View File
@@ -7,11 +7,9 @@ Covers:
- TPH-#63: Horizon anchoring at evidence endpoint
- TPH-#64: Burst/habit checks with unknown daily shares
"""
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"))
@@ -19,8 +17,8 @@ from fenris.store import init_store
from fenris.monitoring_periods import ensure_period_open, close_period
from fenris.derive import derive_hours_from_interval
from fenris.day_aggregate import derive_day
from fenris.projection import compute_projection, ConfidenceState, STALENESS_HOURS
from fenris.tui import _query_live_graph_data, _query_daily_graph_data
from fenris.projection import compute_projection
from fenris.tui import _query_live_graph_data
# ---------------------------------------------------------------------------
+2 -4
View File
@@ -6,7 +6,6 @@ These tests verify that:
3. Edge cases like zero-delta intervals and unknown daily shares are handled
4. qualifying_days_progress shows honest qualifying day count
"""
import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path
@@ -15,10 +14,9 @@ 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, close_period
from fenris.monitoring_periods import ensure_period_open
from fenris.projection import (
compute_projection, ConfidenceState, BaselineTier,
WARMING_MIN_DAYS, WARMING_COVERAGE_FLOOR, WARMING_MAX_LOW_COVERAGE,
compute_projection, ConfidenceState,
)
+18 -19
View File
@@ -5,7 +5,6 @@ Covers:
- TPH-11: Single maker-credit placement, vendor wear under Drive health
- Preserve: continuity, pause block, quit rail, auth banner
"""
import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch
@@ -93,8 +92,8 @@ class TestFenrisIdentity:
store_path=tmp_path / "nonexistent.db",
refresh_interval_s=0.2,
)
async with app.run_test(size=(120, 24)) as pilot:
headline = str(app.query_one("#headline-band").render())
async with app.run_test(size=(120, 24)):
str(app.query_one("#headline-band").render())
assert "🐺 Fenris by Bongbetic" in str(app.query_one("#identity").render())
@pytest.mark.asyncio
@@ -105,8 +104,8 @@ class TestFenrisIdentity:
refresh_interval_s=0.2,
)
# Simulate narrow terminal that can't render wolf
async with app.run_test(size=(60, 24)) as pilot:
headline = str(app.query_one("#headline-band").render())
async with app.run_test(size=(60, 24)):
str(app.query_one("#headline-band").render())
# Either shows wolf or fallback - both are acceptable
assert "Fenris by Bongbetic" in str(app.query_one("#identity").render())
@@ -117,7 +116,7 @@ class TestFenrisIdentity:
store_path=tmp_path / "nonexistent.db",
refresh_interval_s=0.2,
)
async with app.run_test(size=(120, 24)) as pilot:
async with app.run_test(size=(120, 24)):
headline = str(app.query_one("#headline-band").render())
# No replacement character (U+FFFD) should appear
assert "\ufffd" not in headline.lower()
@@ -137,7 +136,7 @@ class TestFenrisIdentity:
conn.close()
app = FenrisTuiApp(store_path=tmp_path / "test.db")
async with app.run_test(size=(120, 24)) as pilot:
async with app.run_test(size=(120, 24)):
headline = str(app.query_one("#headline-band").render())
# Identity appears once, lifespan is separate data
assert "🐺 Fenris by Bongbetic" in str(app.query_one("#identity").render())
@@ -160,7 +159,7 @@ class TestSingleMakerCredit:
conn.close()
app = FenrisTuiApp(store_path=tmp_path / "test.db")
async with app.run_test(size=(120, 24)) as pilot:
async with app.run_test(size=(120, 24)):
strip = str(app.query_one("#service-strip").render())
# "by Bongbetic" should NOT appear in service strip
assert "by Bongbetic" not in strip
@@ -172,15 +171,15 @@ class TestSingleMakerCredit:
store_path=tmp_path / "nonexistent.db",
refresh_interval_s=0.2,
)
async with app.run_test(size=(120, 24)) as pilot:
headline = str(app.query_one("#headline-band").render())
async with app.run_test(size=(120, 24)):
str(app.query_one("#headline-band").render())
assert "Fenris by Bongbetic" in str(app.query_one("#identity").render())
@pytest.mark.asyncio
async def test_empty_store_no_maker_credit_in_strip(self, tmp_path):
"""Empty store: no maker credit in service strip."""
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app.run_test(size=(120, 24)) as pilot:
async with app.run_test(size=(120, 24)):
strip = str(app.query_one("#service-strip").render())
assert "by Bongbetic" not in strip
@@ -191,7 +190,7 @@ class TestSingleMakerCredit:
db = tmp_path / "test.db"
db.write_bytes(b"not a database")
app = FenrisTuiApp(store_path=db)
async with app.run_test(size=(120, 24)) as pilot:
async with app.run_test(size=(120, 24)):
strip = str(app.query_one("#service-strip").render())
assert "by Bongbetic" not in strip
@@ -212,7 +211,7 @@ class TestDriveHealthVendorWear:
conn.close()
app = FenrisTuiApp(store_path=tmp_path / "test.db")
async with app.run_test(size=(120, 24)) as pilot:
async with app.run_test(size=(120, 24)):
health = str(app.query_one("#drive-health").render())
# Vendor wear should be present with health context
assert "vendor wear" in health.lower()
@@ -243,7 +242,7 @@ class TestDriveHealthVendorWear:
conn.close()
app = FenrisTuiApp(store_path=tmp_path / "test.db")
async with app.run_test(size=(120, 24)) as pilot:
async with app.run_test(size=(120, 24)):
health = str(app.query_one("#drive-health").render())
# Should show 0% used or honest zero, not crash
assert "vendor wear" in health.lower()
@@ -265,7 +264,7 @@ class TestPreservedBehavior:
"last_collect_ok": True, "last_collect_age_s": 60,
"last_collect_reason": None,
}):
async with app.run_test(size=(120, 24)) as pilot:
async with app.run_test(size=(120, 24)):
strip = str(app.query_one("#service-strip").render())
assert "continuity" in strip.lower()
assert "monitoring" in strip.lower()
@@ -274,7 +273,7 @@ class TestPreservedBehavior:
async def test_quit_rail_preserved(self, tmp_path):
"""Separate q Quit TUI rail preserved."""
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app.run_test(size=(120, 24)) as pilot:
async with app.run_test(size=(120, 24)):
rail = str(app.query_one("#action-rail").render())
assert "q Quit TUI" in rail
@@ -286,11 +285,11 @@ class TestPreservedBehavior:
refresh_interval_s=0.2,
)
async with app.run_test(size=(120, 24)) as pilot:
headline = str(app.query_one("#headline-band").render())
str(app.query_one("#headline-band").render())
assert "polkit" in str(app.query_one("#auth-notice").render()).lower()
# Should clear after first tick
await pilot.pause(0.25)
headline_after = str(app.query_one("#headline-band").render())
str(app.query_one("#headline-band").render())
assert not app.query_one("#auth-notice").display
@pytest.mark.asyncio
@@ -312,7 +311,7 @@ class TestPreservedBehavior:
"last_collect_ok": None, "last_collect_age_s": None,
"last_collect_reason": None,
}):
async with app.run_test(size=(120, 24)) as pilot:
async with app.run_test(size=(120, 24)):
banner = str(app.query_one("#paused-banner").render())
assert "paused" in banner.lower()
assert "deliberate" in banner.lower()
+10 -15
View File
@@ -8,7 +8,6 @@ Covers:
- AC80-5: Theme roles for graph rendering
- AC80-6: Headless interaction tests with temporary user config
"""
import json
import os
from datetime import datetime, timedelta, timezone
from pathlib import Path
@@ -21,15 +20,12 @@ sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.store import init_store
from fenris.preferences import (
load_preferences,
save_preferences,
get_preference_path,
)
from fenris.themes import get_theme, get_graph_colors, THEME_NAMES
from fenris.themes import get_graph_colors, THEME_NAMES
from fenris.status_composition import (
StatusState,
compose_status,
render_status_tui,
)
@@ -100,7 +96,7 @@ class TestPresetLoading:
config_home = _make_prefs_dir(tmp_path)
with patch.dict(os.environ, {"XDG_CONFIG_HOME": str(config_home)}):
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app.run_test() as pilot:
async with app.run_test():
# Theme should be fenris-amber
assert app.theme == "fenris-chalktone"
@@ -113,7 +109,7 @@ class TestPresetLoading:
with patch.dict(os.environ, {"XDG_CONFIG_HOME": str(config_home)}):
save_preferences(theme="nord", reduced_motion=False)
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app.run_test() as pilot:
async with app.run_test():
assert app.theme == "fenris-nord"
@pytest.mark.asyncio
@@ -125,20 +121,19 @@ class TestPresetLoading:
with patch.dict(os.environ, {"XDG_CONFIG_HOME": str(config_home)}):
save_preferences(theme="high_contrast", reduced_motion=False)
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app.run_test() as pilot:
async with app.run_test():
assert app.theme == "fenris-high-contrast"
@pytest.mark.asyncio
async def test_tui_applies_reduced_motion_from_prefs(self, tmp_path):
"""TUI respects reduced_motion preference."""
from fenris.tui import FenrisTuiApp
from fenris.status_composition import compose_status
config_home = _make_prefs_dir(tmp_path)
with patch.dict(os.environ, {"XDG_CONFIG_HOME": str(config_home)}):
save_preferences(theme="amber", reduced_motion=True)
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app.run_test() as pilot:
async with app.run_test():
# The app should have reduced_motion set
assert app._reduced_motion is True
@@ -237,7 +232,7 @@ class TestPersistence:
with patch.dict(os.environ, {"XDG_CONFIG_HOME": str(config_home)}):
# First run: change theme via preferences API
app1 = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app1.run_test() as pilot:
async with app1.run_test():
save_preferences(theme="nord", reduced_motion=False)
app1._current_theme_name = "nord"
app1.theme = "fenris-nord"
@@ -246,7 +241,7 @@ class TestPersistence:
# Second run: theme should persist
app2 = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app2.run_test() as pilot:
async with app2.run_test():
assert app2.theme == theme_after
@pytest.mark.asyncio
@@ -334,7 +329,7 @@ class TestReducedMotion:
config_home = _make_prefs_dir(tmp_path)
with patch.dict(os.environ, {"XDG_CONFIG_HOME": str(config_home)}):
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app.run_test() as pilot:
async with app.run_test():
# The app should check for reduced motion on mount
assert hasattr(app, '_reduced_motion')
@@ -380,7 +375,7 @@ class TestSafePersistence:
with patch.dict(os.environ, {"XDG_CONFIG_HOME": str(config_home)}):
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app.run_test() as pilot:
async with app.run_test():
# TUI should start with default theme
assert app.theme == "fenris-chalktone"
# Dashboard should be functional
@@ -399,7 +394,7 @@ class TestSafePersistence:
with patch.dict(os.environ, {"XDG_CONFIG_HOME": str(config_home)}):
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app.run_test() as pilot:
async with app.run_test():
# TUI should start without crash
assert app.theme in ("fenris-chalktone", "fenris-amber", "fenris-nord", "fenris-high-contrast")
-2
View File
@@ -10,7 +10,6 @@ Covers:
- AC92-7: Keyboard actions verified from normal launch with headless driver
- AC92-8: Equivalent clickable actions at 80x24 and constrained widths
"""
import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch
@@ -111,7 +110,6 @@ def _make_app(tmp_path, clock=None):
"""
if clock is None:
clock = _clock()
from unittest.mock import patch
import fenris.tui as tui_mod
from datetime import datetime as _real_datetime
+3 -7
View File
@@ -11,7 +11,6 @@ Seams:
- Read side: query_local_day_history() returns entries with evidence flags
- Repair: repair_derivation() does not touch existing local_days
"""
import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Dict
@@ -21,7 +20,7 @@ import sys
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.store import init_store, SCHEMA_VERSION
from fenris.store import init_store
from fenris.local_day import (
derive_local_day_summary,
persist_local_day,
@@ -29,11 +28,9 @@ from fenris.local_day import (
query_local_day_history,
LocalDaySummary,
LocalDayHistoryEntry,
_is_detail_available,
)
from fenris.pruning import prune_old_samples, RAW_SAMPLE_RETENTION_DAYS
from fenris.pruning import prune_old_samples
from fenris.repair import repair_derivation
from fenris.monitoring_periods import ensure_period_open, close_period
# ---------------------------------------------------------------------------
@@ -226,7 +223,7 @@ class TestBoundaryPreservation:
_insert_sample(conn, "2026-09-16T23:55:00+00:00", bw=100)
_insert_sample(conn, "2026-09-17T12:30:00+00:00", bw=200)
pruned = prune_old_samples(conn, now, retention_days=14)
prune_old_samples(conn, now, retention_days=14)
# Boundary anchor should be retained
cursor = conn.execute(
@@ -305,7 +302,6 @@ class TestLegacyDataHandling:
def test_incomplete_legacy_labelled_correctly(self, tmp_path):
"""Legacy summary without full evidence is labelled incomplete."""
conn = init_store(tmp_path / "obs.db")
now = datetime(2026, 10, 1, 12, 0, 0, tzinfo=timezone.utc)
# Simulate a legacy summary that can't establish a full local-day total
legacy = LocalDaySummary(
+1 -4
View File
@@ -4,9 +4,6 @@ Tests the idempotent, interruption-safe import of history.jsonl into the
observation store.
"""
import json
import os
import sqlite3
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict
@@ -17,7 +14,7 @@ import pytest
import sys
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.legacy import import_legacy_history, is_legacy_imported, _parse_history_line
from fenris.legacy import import_legacy_history, _parse_history_line
from fenris.store import init_store
-4
View File
@@ -21,9 +21,7 @@ import sys
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.collector import run_collection
from fenris.store import init_store
from fenris.day_aggregate import derive_day
from fenris.monitoring_periods import ensure_period_open
from fenris.tui import _query_daily_graph_data, _query_hourly_graph_data
@@ -247,7 +245,6 @@ class TestCrossHourAttribution:
for day in ("2026-09-01",):
agg = derive_day(conn, day)
assert agg is not None
total_accounted = agg.bytes_written_delta + agg.bytes_read_delta
unattributed_w = conn.execute(
"SELECT unattributed_bytes_written FROM day_aggregates WHERE day = ?",
(day,)
@@ -483,7 +480,6 @@ class TestCrossDayBoundary:
# The cross-day delta is unattributed at the day level
sep1 = day_map["2026-09-01"]
sep2 = day_map["2026-09-02"]
# Both days may show the unattributed bytes
# (the delta is added to both days' unattributed totals as evidence)
total_w = sep1["allocated_bytes"] + sep1["unallocated_bytes"]
+6 -7
View File
@@ -3,7 +3,6 @@
Spec: §8.4, §8.5, §8.6, §8.7
"""
import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from unittest.mock import patch, MagicMock
@@ -56,7 +55,7 @@ class TestEnableIdempotentMatrix:
"""A fresh package install has a store directory but no database yet."""
args = MagicMock(now=False, store_path=store_path)
with patch("fenris.monitor.enable_timer") as mock_enable:
with patch("fenris.monitor.enable_timer"):
cmd_enable(args)
conn = init_store(store_path)
@@ -73,7 +72,7 @@ class TestEnableIdempotentMatrix:
args = MagicMock(now=False, store_path=store_path)
with patch("fenris.monitor.enable_timer") as mock_enable:
with patch("fenris.monitor.enable_timer"):
cmd_enable(args)
# Period should be open
@@ -98,7 +97,7 @@ class TestEnableIdempotentMatrix:
args = MagicMock(now=True, store_path=store_path)
with patch("fenris.monitor.enable_timer") as mock_enable:
with patch("fenris.monitor.enable_timer"):
cmd_enable(args)
# Should still have exactly one open period
@@ -122,7 +121,7 @@ class TestEnableIdempotentMatrix:
args = MagicMock(now=True, store_path=store_path)
with patch("fenris.monitor.enable_timer") as mock_enable:
with patch("fenris.monitor.enable_timer"):
cmd_enable(args)
# Should have a new open period
@@ -151,7 +150,7 @@ class TestDisableIdempotentMatrix:
args = MagicMock(now=True, store_path=store_path)
with patch("fenris.monitor.disable_timer") as mock_disable:
with patch("fenris.monitor.disable_timer"):
cmd_disable(args)
# Period should be closed with user_disabled
@@ -169,7 +168,7 @@ class TestDisableIdempotentMatrix:
args = MagicMock(now=True, store_path=store_path)
with patch("fenris.monitor.disable_timer") as mock_disable:
with patch("fenris.monitor.disable_timer"):
cmd_disable(args)
# No periods should exist
+1 -2
View File
@@ -6,9 +6,8 @@ From spec §5.2, §8.6, §9.8:
- Powered-off time stays inside a period; disabled time does not
- End causes: user_disabled, migrated, unknown_gap
"""
import sqlite3
import sys
from datetime import datetime, timezone, timedelta
from datetime import datetime, timezone
from pathlib import Path
import pytest
+1 -5
View File
@@ -10,11 +10,11 @@ Covers:
- CLI status and collector behaviour unchanged by preferences
"""
import json
from datetime import datetime, timezone
import os
from pathlib import Path
from unittest.mock import patch
import pytest
import sys
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
@@ -295,7 +295,3 @@ class TestCLIIsolation:
tables_after = sorted(r[0] for r in cursor.fetchall())
conn.close()
assert tables_before == tables_after
# Need datetime for CLI isolation test
from datetime import datetime, timezone
+2 -6
View File
@@ -9,7 +9,6 @@ Covers acceptance criteria:
- PR-13: Baseline provenance and validation per register
- PR-17: Arithmetic exactly E_rated = TBW * 10^12, E_implied = 100*W/p, projected = max(E-W,0)/rate
"""
import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path
@@ -18,11 +17,10 @@ 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, close_period
from fenris.monitoring_periods import ensure_period_open
from fenris.projection import (
compute_projection, ConfidenceState, BaselineTier, ScenarioRange,
TBW_TO_BYTES, HORIZON_DAYS, WARMING_MIN_DAYS, STALENESS_HOURS,
YOUNG_REGIME_DAYS, DISCLOSURES, _compute_horizon_rate,
TBW_TO_BYTES, DISCLOSURES,
)
@@ -1039,8 +1037,6 @@ class TestEvidenceAnchoredHorizons:
# cumulative_bytes = 30 * 100 * 1024 * 1024
# rate = cumulative_bytes / wall_clock
# headline = (E_baseline - cumulative_bytes) / rate
E_baseline = 10.0 * TBW_TO_BYTES
cumulative_bytes = 30 * bw
assert result.headline_remaining_seconds >= 0
def test_zero_boundary_delta_returns_zero(self, store):
+2 -3
View File
@@ -5,7 +5,6 @@ day aggregates are retained indefinitely.
Issue #74: Boundary anchors required for successor evidence are retained.
"""
import sqlite3
import sys
from datetime import datetime, timezone, timedelta
from pathlib import Path
@@ -124,7 +123,7 @@ class TestPruneOldSamples:
_insert_sample(store_conn, "2026-09-16T12:30:00+00:00", 2000000)
# Run pruning
pruned = prune_old_samples(store_conn, now, retention_days=14)
prune_old_samples(store_conn, now, retention_days=14)
# The boundary anchor should be retained
cursor = store_conn.execute(
@@ -148,7 +147,7 @@ class TestPruneOldSamples:
store_conn.commit()
# Run pruning
pruned = prune_old_samples(store_conn, now, retention_days=14)
prune_old_samples(store_conn, now, retention_days=14)
# Keep the source samples until local-day replacement evidence exists.
cursor = store_conn.execute(
+1 -3
View File
@@ -58,7 +58,7 @@ def _rpm_filename(version: str, release: int = 1) -> str:
def _registry_upload_deb_url(version: str) -> str:
"""Expected registry upload URL for a deb package."""
return f"debian/pool/bookworm/main/upload"
return "debian/pool/bookworm/main/upload"
def _registry_upload_rpm_url() -> str:
@@ -258,8 +258,6 @@ class TestRevisionBumping:
"""When revision is bumped, the deb filename also changes."""
version = _get_version()
# Deb filename includes release in nfpm naming
deb_r1 = f"fenris_{version}_amd64.deb"
deb_r2 = f"fenris_{version}_amd64.deb"
# For deb, the filename doesn't change with revision (deb uses epoch)
# But the RPM does — this verifies we test RPM revision correctly
rpm_r1 = _rpm_filename(version, 1)
+6 -8
View File
@@ -4,9 +4,8 @@ Tests the idempotent, safe repair of hour observations and day aggregates
from surviving raw samples, boundary anchor retention, and legacy summary
handling at actual precision.
"""
import sqlite3
import sys
from datetime import datetime, timedelta, timezone
from datetime import datetime, timezone
from pathlib import Path
import pytest
@@ -102,7 +101,6 @@ class TestRepairUsesSameEvidenceRules:
def test_repair_idempotent_on_fully_derived(self, store_conn):
"""Repair on store with existing derived data does not duplicate."""
now = datetime(2026, 9, 15, 12, 0, 0, tzinfo=timezone.utc)
_open_period(store_conn, "2026-09-14T00:00:00+00:00")
# Insert samples that span an hour
@@ -138,7 +136,7 @@ class TestRepairUsesSameEvidenceRules:
store_conn.commit()
# Run repair
result = repair_derivation(store_conn)
repair_derivation(store_conn)
# Verify marker preserved
cursor = store_conn.execute(
@@ -265,7 +263,7 @@ class TestBoundaryAnchorRetention:
_insert_sample(store_conn, "2026-09-29T12:00:00+00:00", 3000000)
# Run pruning
pruned = prune_old_samples(store_conn, now, retention_days=14)
prune_old_samples(store_conn, now, retention_days=14)
# The boundary anchor should be retained
cursor = store_conn.execute(
@@ -286,7 +284,7 @@ class TestBoundaryAnchorRetention:
bytes_written_delta=1000000)
# Run pruning
pruned = prune_old_samples(store_conn, now, retention_days=14)
prune_old_samples(store_conn, now, retention_days=14)
# The source remains until local-day evidence is also durable.
cursor = store_conn.execute(
@@ -309,7 +307,7 @@ class TestLegacySummaryPrecision:
bytes_written_delta=5000000)
# Run repair
result = repair_derivation(store_conn)
repair_derivation(store_conn)
# Should not create hour observations for legacy day
cursor = store_conn.execute(
@@ -324,7 +322,7 @@ class TestLegacySummaryPrecision:
bytes_written_delta=5000000)
# Run repair
result = repair_derivation(store_conn)
repair_derivation(store_conn)
# Day aggregate should not be modified
cursor = store_conn.execute(
+2 -4
View File
@@ -11,13 +11,11 @@ Covers:
- Retired command rejection with migration pointers
- Configuration error surfaced from direct reads
"""
import os
import sqlite3
import sys
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch, MagicMock
from unittest.mock import patch
import pytest
@@ -35,7 +33,7 @@ from fenris.status import (
ACCURACY_SEC,
)
from fenris.store import init_store, SCHEMA_VERSION
from fenris.projection import DISCLOSURES, ConfidenceState
from fenris.projection import DISCLOSURES
# ---------------------------------------------------------------------------
+4 -9
View File
@@ -10,10 +10,9 @@ Covers:
- AC78-5: Separate freshness, last outcome, boot enablement, activity facts
- AC78-6: 5s poll reevaluates freshness from cached clock
"""
import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch, MagicMock
from unittest.mock import patch
import pytest
import sys
@@ -23,12 +22,9 @@ sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.store import init_store
from fenris.status_composition import (
StatusState,
StatusComposition,
compose_status,
render_status_cli,
render_status_tui,
FRESH_THRESHOLD_S,
STALENESS_THRESHOLD_S,
)
@@ -682,7 +678,6 @@ class TestStoreFaultSuppression:
conn.close()
def test_newer_schema_suppresses_projection(self, tmp_path):
import fenris.store as store_mod
conn = init_store(tmp_path / "test.db")
now = _clock()
@@ -971,7 +966,7 @@ class TestTUIIntegration:
"last_collect_ok": True, "last_collect_age_s": 120,
"last_collect_reason": None,
}):
async with app.run_test() as pilot:
async with app.run_test():
strip = str(app.query_one("#service-strip").render())
assert "●" in strip
assert "Monitoring" in strip or "monitoring" in strip.lower()
@@ -997,7 +992,7 @@ class TestTUIIntegration:
"last_collect_ok": None, "last_collect_age_s": None,
"last_collect_reason": None,
}):
async with app.run_test() as pilot:
async with app.run_test():
strip = str(app.query_one("#service-strip").render())
# Should show Paused state
assert "‖" in strip or "Paused" in strip or "paused" in strip.lower()
@@ -1024,7 +1019,7 @@ class TestTUIIntegration:
"last_collect_ok": False, "last_collect_age_s": 60,
"last_collect_reason": "exit code 3",
}):
async with app.run_test() as pilot:
async with app.run_test():
strip = str(app.query_one("#service-strip").render())
# Should show Error state
assert "✖" in strip or "Error" in strip or "error" in strip.lower()
+1 -2
View File
@@ -6,7 +6,6 @@ instead of degrading to the Store fault view; (2) even group members could
not open the WAL-mode store because root-created sidecars lacked group write
and the store directory lacked group execute-then-write.
"""
import sqlite3
import sys
from pathlib import Path
@@ -14,7 +13,7 @@ import pytest
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.store import DEFAULT_STORE_PATH, init_store
from fenris.store import init_store
from fenris.status import StoreFault, open_store_readonly
-1
View File
@@ -7,7 +7,6 @@ Covers:
- Status semantic colours/glyphs/text always win over theme
- Global action reachability and focus contrast in every preset
"""
import pytest
import sys
from pathlib import Path
+13 -21
View File
@@ -13,29 +13,22 @@ import sqlite3
from xml.etree import ElementTree
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch, MagicMock
from unittest.mock import patch
import pytest
import sys
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from textual.app import App
from textual.pilot import Pilot
from fenris.store import init_store, SCHEMA_VERSION
from fenris.monitoring_periods import ensure_period_open, close_period
from fenris.store import init_store
from fenris.monitoring_periods import ensure_period_open
from fenris.projection import (
ConfidenceState,
compute_projection,
DISCLOSURES,
WARMING_MIN_DAYS,
STALENESS_HOURS,
YOUNG_REGIME_DAYS,
)
from fenris.status import (
FRESH_THRESHOLD_S,
STALENESS_THRESHOLD_S,
grade_freshness,
)
from fenris.tui import (
@@ -45,7 +38,6 @@ from fenris.tui import (
_query_drive_health,
_query_daily_graph_data,
_query_hourly_graph_data,
_RANGE_OPTIONS,
_RANGE_DEFAULT,
_MIN_WIDTH,
_MIN_HEIGHT,
@@ -283,7 +275,7 @@ class TestDenseScreen:
conn.close()
app = FenrisTuiApp(store_path=tmp_path / "test.db")
async with app.run_test() as pilot:
async with app.run_test():
assert app.query_one("#headline-band") is not None
assert app.query_one("#usage-history") is not None
assert app.query_one("#drive-health") is not None
@@ -303,7 +295,7 @@ class TestDenseScreen:
conn.close()
app = FenrisTuiApp(store_path=tmp_path / "test.db")
async with app.run_test() as pilot:
async with app.run_test():
headline = str(app.query_one("#headline-band").render())
assert "remaining" in headline.lower() or "projection" in headline.lower()
@@ -321,7 +313,7 @@ class TestDenseScreen:
conn.close()
app = FenrisTuiApp(store_path=tmp_path / "test.db")
async with app.run_test() as pilot:
async with app.run_test():
headline = str(app.query_one("#headline-band").render())
assert "projection confidence" in headline.lower()
# Contributing facts shown, never a percentage as confidence
@@ -343,7 +335,7 @@ class TestDenseScreen:
conn.close()
app = FenrisTuiApp(store_path=tmp_path / "test.db")
async with app.run_test() as pilot:
async with app.run_test():
strip = str(app.query_one("#service-strip").render())
assert "Boot:" in strip
assert "Timer:" in strip
@@ -484,7 +476,7 @@ class TestDenseScreen:
auth_notice = "Open with fenris (no sudo). Actions authenticate via polkit. ? Help"
async with app.run_test() as pilot:
headline = str(app.query_one("#headline-band").render())
str(app.query_one("#headline-band").render())
# Identity now shows Fenris by Bongbetic (issue #79)
assert "Fenris by Bongbetic" in str(app.query_one("#identity").render())
assert auth_notice in str(app.query_one("#auth-notice").render())
@@ -526,7 +518,7 @@ class TestDisclosuresAndGreeting:
async def test_empty_store_greeting(self, tmp_path):
"""Empty store shows 'no observations yet' with enable hint."""
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app.run_test() as pilot:
async with app.run_test():
headline = str(app.query_one("#headline-band").render())
assert "no observations yet" in headline.lower()
assert "enable" in headline.lower() or "resume" in headline.lower()
@@ -591,9 +583,9 @@ class TestFirstRun:
async def test_first_run_prompt(self, tmp_path):
"""First-run prompt makes the keyboard action and boot effect explicit."""
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app.run_test() as pilot:
async with app.run_test():
headline = str(app.query_one("#headline-band").render())
strip = str(app.query_one("#service-strip").render())
str(app.query_one("#service-strip").render())
assert "no observations yet" in headline.lower()
assert "r resume — enable monitoring and future boots" in headline.lower()
assert "r resume" in str(app.query_one("#action-rail").render()).lower()
@@ -866,7 +858,7 @@ class TestQueryHourlyGraphData:
class TestHistoryGraph:
def test_empty_data(self, tmp_path):
"""Empty data shows awaiting message."""
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
# We test the widget directly via the app's compose
graph = HistoryGraph()
# Simulate setting empty data
@@ -1129,7 +1121,7 @@ class TestBarGraphTUI:
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app.run_test(size=(80, 24)) as pilot:
await pilot.press("v", "v")
graph = app.query_one("#usage-history")
app.query_one("#usage-history")
assert app.query_one("#activity-panel").border_title == "Drive activity"
@pytest.mark.asyncio