235 lines
11 KiB
Python
235 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import httpx
|
|
import pytest
|
|
from sqlmodel import Session, SQLModel, create_engine, select
|
|
|
|
from backend.app.db.models import AuditEvent, JobRun
|
|
from backend.app.services import action_execution, deployment_broker, runner_statistics
|
|
|
|
|
|
def snapshot():
|
|
at = datetime.now(UTC).isoformat()
|
|
return {
|
|
"schemaVersion": 1, "status": "healthy", "scope": "ci-fleet", "generatedAt": at,
|
|
"budget": {"cpuLimit": 8, "memoryLimitBytes": 16 * 1024**3, "activeJobLimit": 2,
|
|
"heavyJobLimit": 1, "reservedCpu": 2, "reservedMemoryBytes": 3 * 1024**3,
|
|
"activeJobs": 0, "heavyJobs": 0},
|
|
"host": {"cpuBusyPercent": None, "loadRatio": None,
|
|
"availableMemoryBytes": None, "cacheFreeBytes": None},
|
|
"profiles": [{"id": "checks", "displayName": "ITWorx CI · Checks",
|
|
"runnerName": "itworx-ci-checks", "labels": ["itworx-ci-checks"],
|
|
"status": "capacity_wait", "reason": "host_headroom",
|
|
"cpuLimit": 1, "memoryLimitBytes": 1536 * 1024**2,
|
|
"queueDepth": 1, "activeJobs": 0, "cpuUsed": None,
|
|
"memoryUsedBytes": None, "oldestQueuedSeconds": 12,
|
|
"completedJobs": 0, "p50WaitSeconds": None, "p95WaitSeconds": None,
|
|
"p50DurationSeconds": None, "p95DurationSeconds": None}],
|
|
"window": {"startAt": at, "endAt": at, "completedJobs": 0, "samples": 1,
|
|
"coverage": "bounded"},
|
|
"history": [{"at": at, "activeJobs": 0, "queuedJobs": 1,
|
|
"cpuUsed": None, "memoryUsedBytes": None}],
|
|
}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_cache(monkeypatch):
|
|
monkeypatch.setattr(runner_statistics, "_cached", None)
|
|
monkeypatch.setattr(runner_statistics, "_cached_at", 0.0)
|
|
|
|
|
|
def test_unknown_measurements_survive_and_private_fields_are_removed():
|
|
payload = snapshot()
|
|
payload["rawJobs"] = [{"secret": "fixture-secret"}]
|
|
payload["profiles"][0]["workspacePath"] = "/private/credentials"
|
|
result = runner_statistics.sanitize_statistics(payload)
|
|
assert result["status"] == "healthy"
|
|
assert result["profiles"][0]["cpuUsed"] is None
|
|
assert result["profiles"][0]["queueDepth"] == 1
|
|
assert "rawJobs" not in result
|
|
assert "workspacePath" not in result["profiles"][0]
|
|
|
|
|
|
def test_unattested_profile_limits_remain_unknown():
|
|
payload = snapshot()
|
|
payload["status"] = "degraded"
|
|
payload["profiles"][0].update(cpuLimit=None, memoryLimitBytes=None)
|
|
result = runner_statistics.sanitize_statistics(payload)
|
|
assert result["status"] == "degraded"
|
|
assert result["profiles"][0]["cpuLimit"] is None
|
|
assert result["profiles"][0]["memoryLimitBytes"] is None
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", [
|
|
lambda p: p.update(schemaVersion=2),
|
|
lambda p: p.update(generatedAt="2026-10-01T12:00:00"),
|
|
lambda p: p.update(generatedAt=(datetime.now(UTC) + timedelta(minutes=1)).isoformat()),
|
|
lambda p: p["profiles"].append(copy.deepcopy(p["profiles"][0])),
|
|
lambda p: p["profiles"][0].update(cpuUsed=float("nan")),
|
|
lambda p: p["profiles"][0].update(activeJobs=True),
|
|
lambda p: p["profiles"][0].update(runnerName="secret-token"),
|
|
lambda p: p["profiles"][0].update(labels=["security-validation"]),
|
|
lambda p: p["profiles"][0].update(reason="Bearer fixture-secret-token"),
|
|
lambda p: p["window"].update(startAt=None),
|
|
lambda p: p["history"].append(copy.deepcopy(p["history"][0])),
|
|
lambda p: p.update(history=[p["history"][0]] * 289),
|
|
])
|
|
def test_untrusted_or_ambiguous_statistics_fail_to_honest_unknown(mutation):
|
|
payload = snapshot()
|
|
mutation(payload)
|
|
result = runner_statistics.sanitize_statistics(payload)
|
|
assert result["status"] == "unavailable"
|
|
assert result["budget"]["cpuLimit"] is None
|
|
assert result["profiles"] == []
|
|
|
|
|
|
def test_old_sample_is_degraded_without_retimestamping():
|
|
payload = snapshot()
|
|
old = (datetime.now(UTC) - timedelta(minutes=2)).isoformat()
|
|
payload["generatedAt"] = old
|
|
payload["history"][0]["at"] = old
|
|
payload["window"].update(startAt=old, endAt=old)
|
|
result = runner_statistics.sanitize_statistics(payload)
|
|
assert result["status"] == "degraded"
|
|
assert result["generatedAt"] == old
|
|
|
|
|
|
def test_singleflight_cache_and_caller_mutation_isolation(monkeypatch):
|
|
calls = []
|
|
|
|
class Broker:
|
|
def is_configured(self):
|
|
return True
|
|
|
|
def runner_fleet_statistics(self):
|
|
calls.append(1)
|
|
return snapshot()
|
|
|
|
monkeypatch.setattr(runner_statistics, "DeploymentBrokerClient", Broker)
|
|
service = runner_statistics.RunnerStatisticsService()
|
|
with ThreadPoolExecutor(max_workers=8) as pool:
|
|
results = list(pool.map(lambda _: service.snapshot(), range(8)))
|
|
assert calls == [1]
|
|
results[0]["budget"]["cpuLimit"] = 999
|
|
assert service.snapshot()["budget"]["cpuLimit"] == 8
|
|
|
|
|
|
@pytest.mark.parametrize("configured,raises,expected", [
|
|
(False, False, "not_configured"), (True, True, "unavailable"),
|
|
])
|
|
def test_unavailable_backend_never_leaks_error_or_invents_budget(monkeypatch, configured, raises, expected):
|
|
class Broker:
|
|
def is_configured(self):
|
|
return configured
|
|
|
|
def runner_fleet_statistics(self):
|
|
assert raises
|
|
raise RuntimeError("Bearer fixture-private-secret /private/key")
|
|
|
|
monkeypatch.setattr(runner_statistics, "DeploymentBrokerClient", Broker)
|
|
result = runner_statistics.RunnerStatisticsService().snapshot()
|
|
assert result["status"] == expected
|
|
assert result["budget"]["reservedCpu"] is None
|
|
assert "fixture-private" not in str(result)
|
|
|
|
|
|
def test_broker_statistics_fixed_read_path_has_deadline_and_no_credentials(monkeypatch):
|
|
calls = []
|
|
real_client = httpx.Client
|
|
|
|
def handle(request):
|
|
calls.append(request)
|
|
return httpx.Response(200, json=snapshot())
|
|
|
|
def client(**kwargs):
|
|
assert kwargs == {"timeout": 5, "verify": True, "follow_redirects": False}
|
|
return real_client(**kwargs, transport=httpx.MockTransport(handle))
|
|
|
|
monkeypatch.setattr(deployment_broker.httpx, "Client", client)
|
|
broker = deployment_broker.DeploymentBrokerClient(base_url="https://controller.test", verify_tls=True)
|
|
assert broker.runner_fleet_statistics()["schemaVersion"] == 1
|
|
request = calls[0]
|
|
assert request.method == "GET"
|
|
assert str(request.url) == "https://controller.test/v1/runner-fleet/statistics"
|
|
assert "authorization" not in request.headers
|
|
assert not any(key.startswith("x-itworx") for key in request.headers)
|
|
assert broker._runner_fleet_read("/v1/arbitrary")["reason"] == "runner_fleet_path_invalid"
|
|
assert len(calls) == 1
|
|
|
|
|
|
def test_specialize_broker_caller_cannot_choose_script_revision_or_target(monkeypatch):
|
|
calls = []
|
|
broker = deployment_broker.DeploymentBrokerClient()
|
|
monkeypatch.setattr(broker, "_request", lambda *args, **kwargs: calls.append((args, kwargs)) or {})
|
|
broker.runner_fleet_specialize(reason="x" * 1100)
|
|
args, kwargs = calls[0]
|
|
assert args == ("POST", "/v1/appops/runner-fleet/specialize")
|
|
assert kwargs["payload"] == {"reason": "x" * 1000}
|
|
|
|
|
|
@pytest.mark.parametrize("status,success", [("specialized", True), ("already_applied", True),
|
|
("upgraded", False), ("failed", False)])
|
|
def test_specialize_execution_persists_actual_success_or_failure(tmp_path, monkeypatch, status, success):
|
|
engine = create_engine(f"sqlite:///{tmp_path / 'specialize.db'}")
|
|
SQLModel.metadata.create_all(engine)
|
|
monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "runner_fleet_specialize",
|
|
lambda self, **kwargs: {"status": status, "revision": "a" * 40})
|
|
with Session(engine) as session:
|
|
result = action_execution.execute_governed_action(
|
|
session, action="runner.fleet.specialize", target="tower/runner-fleet-specialized",
|
|
source="chatgpt-mcp", risk="high", reason="approved fleet specialization",
|
|
original_policy_decision="allowed",
|
|
)
|
|
job = session.get(JobRun, result["jobId"])
|
|
audit = session.get(AuditEvent, result["auditId"])
|
|
assert result["success"] is success
|
|
assert job.status == ("succeeded" if success else "failed")
|
|
assert audit.result == ("success" if success else "failed")
|
|
assert audit.target == "tower/runner-fleet-specialized"
|
|
assert ("verified" in result["message"]) is success
|
|
|
|
|
|
def test_wrong_specialization_target_is_blocked_before_job_or_broker(tmp_path, monkeypatch):
|
|
engine = create_engine(f"sqlite:///{tmp_path / 'blocked.db'}")
|
|
SQLModel.metadata.create_all(engine)
|
|
monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "runner_fleet_specialize",
|
|
lambda *args, **kwargs: pytest.fail("must not call broker"))
|
|
with Session(engine) as session:
|
|
result = action_execution.execute_governed_action(
|
|
session, action="runner.fleet.specialize", target="other/runner",
|
|
source="chatgpt-mcp", risk="high", reason="wrong target",
|
|
original_policy_decision="allowed",
|
|
)
|
|
assert result["policyDecision"] == "blocked"
|
|
assert session.exec(select(JobRun)).all() == []
|
|
|
|
|
|
@pytest.mark.parametrize('status,success', [('configured', True), ('already_configured', True),
|
|
('specialized', False), ('failed', False)])
|
|
def test_statistics_configuration_persists_actual_status_without_secret(tmp_path, monkeypatch, status, success):
|
|
engine = create_engine(f"sqlite:///{tmp_path / 'binding.db'}")
|
|
SQLModel.metadata.create_all(engine)
|
|
monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, 'runner_statistics_configure',
|
|
lambda self, **kwargs: {'status': status, 'revision': 'a' * 40, 'projectbrainDeliveryRequired': True})
|
|
with Session(engine) as session:
|
|
result = action_execution.execute_governed_action(
|
|
session, action='runner.statistics.configure', target='tower/projectbrain-runner-statistics',
|
|
source='chatgpt-mcp', risk='high', reason='configure read-only metrics', original_policy_decision='allowed')
|
|
assert result['success'] is success
|
|
assert session.get(JobRun, result['jobId']).status == ('succeeded' if success else 'failed')
|
|
assert session.get(AuditEvent, result['auditId']).target == 'tower/projectbrain-runner-statistics'
|
|
assert 'token' not in result
|
|
|
|
|
|
def test_statistics_configuration_broker_has_fixed_signed_body(monkeypatch):
|
|
calls = []
|
|
broker = deployment_broker.DeploymentBrokerClient()
|
|
monkeypatch.setattr(broker, '_request', lambda *args, **kwargs: calls.append((args, kwargs)) or {})
|
|
broker.runner_statistics_configure(reason='configure read-only stats')
|
|
assert calls[0][0] == ('POST', '/v1/appops/runner-statistics/configure')
|
|
assert calls[0][1]['payload'] == {'reason': 'configure read-only stats'}
|