Files
roboco/tests/unit/test_config_properties.py
a036c97985 fix(tg): cockpit data correctness — real GLM pricing, display timezone, agent activity tracking (#666)
* fix(tg): cockpit data correctness — real GLM pricing, display timezone, agent activity tracking

Three root causes behind the Mini App/bot showing wrong numbers:

Pricing: glm-5.2 gets a grounded per-token rate (z.ai published pricing,
$1.40/$4.40/$0.26 per 1M, source+date in the table comment) so a GLM
fleet day stops reporting $0.00 for half a million tokens; ungrounded
Ollama-Cloud models render "subscription (untracked)" instead of a bare
zero (is_ollama_cloud_model, consumed directly by the cockpit). Side
effect, intended and documented: honestly-priced GLM now trips the
downgrade-only comparator for new qa/documenter complexity pins.

Display timezone: the cockpit bucketed days in UTC for a GMT+2 operator.
New pure foundation module display_time (resolve_zone/local_date/
trailing_dates/day_bounds_utc, DST-correct with tests for the 23h/25h
days) + ROBOCO_DISPLAY_TIMEZONE (IANA-validated, default UTC); the
cockpit's spend/velocity series bucket raw session/completion rows by
the display zone. The UTC-keyed rollup table and the main dashboard are
deliberately untouched.

Agent activity: AgentTable.status was never set to ACTIVE and
current_task_id was never written anywhere — "active: 0, working: []"
was structurally permanent. Every claim path now marks the claimant
ACTIVE with rollback symmetry (_finalize_claim for dev/PM claims,
_qa_or_doc_claim for QA/doc/PR-gate claims, pr_review_claim for external
review) and every release path clears it (pass/fail QA, pr_pass/pr_fail,
complete_review, advance-to-PM-review, reaper unclaim, voluntary
unclaim, reassign retarget, pool divert, admin transitions, unblock
restore-to-in-progress). The bot's /status shares the cockpit's fleet
derivation so the two surfaces can't disagree. Known ceiling, commented:
one current_task_id column shows a multi-root coordinator PM's most
recent claim only.

Drill: sonnet develop -> sonnet adversarial (refuted the original
chokepoint coverage claim; QA/doc/reviewer paths were unwired) ->
correction round (wired them all + restored a dropped assertion, deleted
a dead helper and the dead subscription_billed field) -> review.

* fix(db): post_update on AgentTable.current_task breaks the flush cycle

agents.current_task_id and tasks.assigned_to reference each other, so a
flush touching both rows — every claim now marks its agent ACTIVE — is
an instance-level circular dependency SQLAlchemy cannot topologically
sort. The e2e smoke's full verb paths (12 tests) hit it; the unit and
integration suites never flush both dirty rows with relationships
loaded. post_update emits the FK as a second UPDATE, the canonical fix
for mutually-referencing rows.

* fix(budgets): enforce only explicitly-set budgets — no per-TaskType defaults

The per-TaskType default cap table blocked an unbudgeted coordination
root one opus planning turn in ($1.50 PLANNING default vs. real
coordination spend) — a false positive by design the moment the fleet
runs a priced model. Budgets are now explicit-input only:
effective_task_budget_usd returns None for an unset budget_usd, the
budget sweep skips enforcement (and never prices spend) on None, and
the unblock re-check passes on None so clearing the budget field is
itself a valid resolution. The project monthly cap stays as the
explicit-input fleet-wide backstop. Panel copy tells the truth
("No cap" placeholder; empty = uncapped), and the TaskType default
table plus its resolver are deleted.

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>
2026-07-23 21:09:50 +02:00

224 lines
8.0 KiB
Python

"""Coverage for roboco.config computed properties."""
from __future__ import annotations
import importlib
import pytest
from pydantic import ValidationError
from roboco.config import Settings, resolve_uvicorn_loop_factory
def test_internal_api_url_uses_api_url_when_set() -> None:
s = Settings(api_url="http://orchestrator:8000")
assert s.internal_api_url == "http://orchestrator:8000/api"
def test_internal_api_url_strips_trailing_slash_on_api_url() -> None:
s = Settings(api_url="http://orchestrator:8000/")
assert s.internal_api_url == "http://orchestrator:8000/api"
def test_internal_api_url_uses_host_when_api_url_unset() -> None:
s = Settings(api_url=None, host="localhost", port=9000)
assert s.internal_api_url == "http://localhost:9000/api"
def test_internal_api_url_swaps_bind_all_to_localhost() -> None:
"""Line 70: host='0.0.0.0' becomes '127.0.0.1' for connecting."""
s = Settings(api_url=None, host="0.0.0.0", port=8000)
assert s.internal_api_url == "http://127.0.0.1:8000/api"
def test_redis_url_with_password_includes_credential() -> None:
"""Line 118: password present → includes :pw@ in URL."""
s = Settings(
redis_host="redis", redis_port=6379, redis_db=2, redis_password="secret"
)
assert s.redis_url == "redis://:secret@redis:6379/2"
def test_redis_url_without_password() -> None:
s = Settings(redis_host="redis", redis_port=6379, redis_db=0, redis_password=None)
assert s.redis_url == "redis://redis:6379/0"
# ---------------------------------------------------------------------------
# Cloud auth — fail-loud secret validation
# ---------------------------------------------------------------------------
def test_cloud_auth_off_does_not_require_secret() -> None:
"""Default (off) construction never raises, secret or not."""
s = Settings(cloud_auth_enabled=False, cloud_auth_secret=None)
assert s.cloud_auth_enabled is False
def test_cloud_auth_enabled_without_secret_fails_loud() -> None:
"""Arming cloud auth with no session-signing secret must fail at startup,
not silently mint unsigned/unsafe sessions."""
with pytest.raises(ValueError, match="ROBOCO_CLOUD_AUTH_SECRET"):
Settings(cloud_auth_enabled=True, cloud_auth_secret=None)
def test_cloud_auth_enabled_with_secret_succeeds() -> None:
s = Settings(cloud_auth_enabled=True, cloud_auth_secret="s" * 32)
assert s.cloud_auth_enabled is True
assert s.cloud_auth_secret == "s" * 32
def test_cloud_auth_cookie_max_age_defaults_to_30_days() -> None:
s = Settings()
assert s.cloud_auth_cookie_max_age == 30 * 24 * 60 * 60
def test_cloud_auth_rejects_panel_agent_token(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("ROBOCO_CLOUD_AUTH_ENABLED", "true")
monkeypatch.setenv("ROBOCO_CLOUD_AUTH_SECRET", "x" * 32)
monkeypatch.setenv("ROBOCO_PANEL_AGENT_TOKEN", "some-signed-token")
with pytest.raises(ValueError, match="ROBOCO_PANEL_AGENT_TOKEN"):
Settings()
def test_cloud_auth_ok_without_panel_agent_token(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("ROBOCO_CLOUD_AUTH_ENABLED", "true")
monkeypatch.setenv("ROBOCO_CLOUD_AUTH_SECRET", "x" * 32)
monkeypatch.delenv("ROBOCO_PANEL_AGENT_TOKEN", raising=False)
s = Settings()
assert s.cloud_auth_enabled is True
# ---------------------------------------------------------------------------
# Telegram Mini App sign-in — requires cloud_auth_enabled
# ---------------------------------------------------------------------------
def test_telegram_miniapp_off_does_not_require_cloud_auth() -> None:
"""Default (off) construction never raises regardless of cloud auth."""
s = Settings(telegram_miniapp_enabled=False, cloud_auth_enabled=False)
assert s.telegram_miniapp_enabled is False
def test_telegram_miniapp_enabled_without_cloud_auth_fails_loud() -> None:
"""The Mini App route mints a cloud-auth session cookie — with cloud
auth off there's nothing to mint, so this must fail at startup."""
with pytest.raises(ValueError, match="ROBOCO_TELEGRAM_MINIAPP_ENABLED"):
Settings(telegram_miniapp_enabled=True, cloud_auth_enabled=False)
def test_telegram_miniapp_enabled_with_cloud_auth_succeeds() -> None:
s = Settings(
telegram_miniapp_enabled=True,
cloud_auth_enabled=True,
cloud_auth_secret="s" * 32,
)
assert s.telegram_miniapp_enabled is True
def test_telegram_initdata_max_age_defaults_to_600() -> None:
ten_minutes = 10 * 60
s = Settings()
assert s.telegram_initdata_max_age_seconds == ten_minutes
# ---------------------------------------------------------------------------
# local_llm_base_url — internal-host guard (H13)
# ---------------------------------------------------------------------------
def test_local_llm_base_url_default_accepted() -> None:
s = Settings()
assert s.local_llm_base_url == "http://roboco-ollama:11434/v1"
def test_local_llm_base_url_localhost_accepted() -> None:
s = Settings(local_llm_base_url="http://localhost:11434")
assert s.local_llm_base_url == "http://localhost:11434"
def test_local_llm_base_url_rfc1918_accepted() -> None:
s = Settings(local_llm_base_url="http://10.0.0.5:11434")
assert s.local_llm_base_url == "http://10.0.0.5:11434"
def test_local_llm_base_url_ipv6_loopback_accepted() -> None:
s = Settings(local_llm_base_url="http://[::1]:11434")
assert s.local_llm_base_url == "http://[::1]:11434"
def test_local_llm_base_url_cluster_local_accepted() -> None:
s = Settings(local_llm_base_url="http://ollama.default.svc.cluster.local:11434")
assert s.local_llm_base_url == "http://ollama.default.svc.cluster.local:11434"
def test_local_llm_base_url_public_rejected() -> None:
with pytest.raises(ValidationError):
Settings(local_llm_base_url="https://api.openai.com/v1")
def test_local_llm_base_url_missing_host_rejected() -> None:
with pytest.raises(ValidationError):
Settings(local_llm_base_url="http://")
# ---------------------------------------------------------------------------
# uvicorn_loop — default asyncio, uvloop opt-in (CI segfault fix)
# ---------------------------------------------------------------------------
def test_uvicorn_loop_defaults_to_asyncio() -> None:
assert Settings().uvicorn_loop == "asyncio"
def test_uvicorn_loop_honors_constructor_override() -> None:
assert Settings(uvicorn_loop="uvloop").uvicorn_loop == "uvloop"
def test_uvicorn_loop_honors_env_override(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("ROBOCO_UVICORN_LOOP", "uvloop")
assert Settings().uvicorn_loop == "uvloop"
def test_uvicorn_loop_rejects_unknown_value() -> None:
with pytest.raises(ValidationError):
Settings(uvicorn_loop="unknown") # type: ignore[arg-type]
def test_resolve_uvicorn_loop_factory_asyncio_is_none() -> None:
"""The default: no override, so asyncio.run() picks its own stock loop."""
assert resolve_uvicorn_loop_factory("asyncio") is None
def test_resolve_uvicorn_loop_factory_uvloop_returns_new_event_loop() -> None:
factory = resolve_uvicorn_loop_factory("uvloop")
assert factory is not None
loop = factory()
try:
uvloop = importlib.import_module("uvloop")
assert isinstance(loop, uvloop.Loop)
finally:
loop.close()
# ---------------------------------------------------------------------------
# display_timezone — the TG cockpit's day-bucketing timezone (Issue 2)
# ---------------------------------------------------------------------------
def test_display_timezone_defaults_to_utc() -> None:
"""Default is a no-op for every deployment that doesn't set it."""
assert Settings().display_timezone == "UTC"
def test_display_timezone_accepts_valid_iana_name() -> None:
assert Settings(display_timezone="Europe/Berlin").display_timezone == (
"Europe/Berlin"
)
def test_display_timezone_rejects_unknown_name() -> None:
with pytest.raises(ValidationError):
Settings(display_timezone="Not/AZone")