193 lines
9.8 KiB
Python
193 lines
9.8 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.testclient import TestClient
|
|
from pydantic import ValidationError
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.admin.models import SystemSettings
|
|
from govoplan_core.auth import ApiPrincipal, get_api_principal
|
|
from govoplan_core.db.session import get_session
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.change_sequence import ChangeSequenceEntry
|
|
from govoplan_core.core.configuration_safety import classify_configuration_field, plan_configuration_change
|
|
from govoplan_core.tenancy.scope import Tenant
|
|
from govoplan_campaign.backend.delivery_policy import SYNCHRONOUS_SEND_MAX_ENV, effective_synchronous_send_policy
|
|
from govoplan_campaign.backend.routes import delivery_settings as routes
|
|
from govoplan_campaign.backend.router import router as campaign_router
|
|
|
|
|
|
def principal(*scopes, tenant="tenant-a"):
|
|
actor = SimpleNamespace(id="admin-1")
|
|
return ApiPrincipal(principal=PrincipalRef(account_id=actor.id, membership_id=actor.id, tenant_id=tenant, scopes=frozenset(scopes)), user=actor, account=actor)
|
|
|
|
|
|
@pytest.fixture
|
|
def policy(tmp_path, monkeypatch):
|
|
monkeypatch.delenv(SYNCHRONOUS_SEND_MAX_ENV, raising=False)
|
|
engine = create_engine(f"sqlite+pysqlite:///{tmp_path / 'policy.db'}")
|
|
for table in (SystemSettings.__table__, Tenant.__table__, ChangeSequenceEntry.__table__):
|
|
table.create(engine)
|
|
with Session(engine) as session:
|
|
session.add_all([
|
|
SystemSettings(id="global", settings={"unrelated": {"enabled": True}}),
|
|
Tenant(id="tenant-a", slug="a", name="A", settings={"unrelated": "preserve"}),
|
|
Tenant(id="tenant-b", slug="b", name="B", settings={}),
|
|
])
|
|
session.commit()
|
|
with patch.object(routes, "audit_from_principal", autospec=True) as audit:
|
|
yield SimpleNamespace(session=session, engine=engine, audit=audit,
|
|
admin=principal("system:settings:read", "system:settings:write", "admin:policies:read", "admin:policies:write"))
|
|
engine.dispose()
|
|
|
|
|
|
def state(policy, scope="system", actor=None):
|
|
return routes.read_delivery_policy(scope, session=policy.session, principal=actor or policy.admin)
|
|
|
|
|
|
def save(policy, value, scope="system", revision=None, actor=None):
|
|
payload = routes.DeliveryPolicyUpdate(synchronous_send_max_recipients=value, expected_revision=revision or state(policy, scope)["revision"])
|
|
return routes.update_delivery_policy(scope, payload, session=policy.session, principal=actor or policy.admin)
|
|
|
|
|
|
def test_system_admin_can_raise_implicit_default_and_save_is_durable_audited_and_scoped(policy):
|
|
before = state(policy)
|
|
assert before["effective_max_recipients"] == 25 and before["max_configurable_recipients"] == 500
|
|
result = save(policy, 200)
|
|
assert result["effective_max_recipients"] == 200 and result["revision"] != before["revision"]
|
|
with Session(policy.engine) as fresh:
|
|
assert effective_synchronous_send_policy(fresh, tenant_id="tenant-a", environ={}).max_recipient_jobs == 200
|
|
system = fresh.get(SystemSettings, "global")
|
|
assert system.settings["unrelated"] == {"enabled": True}
|
|
history = system.settings["_configuration_control"]["history"]
|
|
assert len(history) == 1 and history[0]["key"] == "campaign_delivery_policy.system"
|
|
assert history[0]["before"]["synchronous_send_max_recipients"] is None
|
|
assert history[0]["after"]["synchronous_send_max_recipients"] == 200
|
|
assert history[0]["actor_user_id"] == "admin-1"
|
|
assert policy.audit.call_args.kwargs["commit"] is False
|
|
assert policy.audit.call_args.kwargs["scope"] == "system"
|
|
|
|
|
|
def test_tenant_can_only_narrow_and_reset_inherits_without_affecting_other_tenant(policy):
|
|
save(policy, 200)
|
|
with pytest.raises(HTTPException) as failure:
|
|
save(policy, 201, "tenant")
|
|
assert failure.value.status_code == 422
|
|
save(policy, 183, "tenant")
|
|
assert effective_synchronous_send_policy(policy.session, tenant_id="tenant-a").max_recipient_jobs == 183
|
|
assert effective_synchronous_send_policy(policy.session, tenant_id="tenant-b").max_recipient_jobs == 200
|
|
assert policy.session.get(Tenant, "tenant-a").settings["unrelated"] == "preserve"
|
|
assert save(policy, None, "tenant")["effective_max_recipients"] == 200
|
|
assert save(policy, None)["effective_max_recipients"] == 25
|
|
|
|
|
|
def test_explicit_deployment_ceiling_remains_authoritative_and_zero_disables(policy, monkeypatch):
|
|
save(policy, 200)
|
|
monkeypatch.setenv(SYNCHRONOUS_SEND_MAX_ENV, "40")
|
|
assert state(policy)["effective_max_recipients"] == 40
|
|
with pytest.raises(HTTPException) as failure:
|
|
save(policy, 41)
|
|
assert failure.value.status_code == 422
|
|
assert save(policy, None)["effective_max_recipients"] == 40
|
|
assert save(policy, 0)["effective_max_recipients"] == 0
|
|
assert state(policy, "tenant")["max_configurable_recipients"] == 0
|
|
|
|
|
|
@pytest.mark.parametrize("blank", ["", " ", "\t\n"])
|
|
def test_blank_deployment_value_never_turns_default_into_absolute_maximum(policy, monkeypatch, blank):
|
|
monkeypatch.setenv(SYNCHRONOUS_SEND_MAX_ENV, blank)
|
|
assert state(policy)["effective_max_recipients"] == 25
|
|
assert state(policy)["deployment_ceiling_explicit"] is False
|
|
save(policy, 200)
|
|
assert state(policy)["effective_max_recipients"] == 200
|
|
save(policy, 183, "tenant")
|
|
assert state(policy, "tenant")["effective_max_recipients"] == 183
|
|
|
|
|
|
def test_stale_parent_own_or_aba_revision_never_overwrites_policy(policy):
|
|
initial = state(policy)
|
|
tenant = state(policy, "tenant")
|
|
save(policy, 200)
|
|
for scope, revision in (("system", initial["revision"]), ("tenant", tenant["revision"])):
|
|
with pytest.raises(HTTPException) as failure:
|
|
save(policy, 10, scope, revision=revision)
|
|
assert failure.value.status_code == 409
|
|
save(policy, None)
|
|
with pytest.raises(HTTPException) as failure:
|
|
save(policy, 30, revision=initial["revision"])
|
|
assert failure.value.status_code == 409
|
|
|
|
|
|
@pytest.mark.parametrize("scope,scopes,operation", [
|
|
("system", ("admin:policies:read", "admin:policies:write"), "read"),
|
|
("system", ("admin:policies:write",), "write"),
|
|
("tenant", ("system:settings:read", "system:settings:write"), "read"),
|
|
("tenant", ("system:settings:write",), "write"),
|
|
("system", ("system:settings:read",), "write"),
|
|
("tenant", ("admin:policies:read",), "write"),
|
|
])
|
|
def test_permissions_cannot_cross_scope_or_use_read_permission_to_write(policy, scope, scopes, operation):
|
|
actor = principal(*scopes)
|
|
with pytest.raises(HTTPException) as failure:
|
|
if operation == "read": state(policy, scope, actor=actor)
|
|
else: save(policy, 10, scope, actor=actor)
|
|
assert failure.value.status_code == 403
|
|
assert "_configuration_control" not in policy.session.get(SystemSettings, "global").settings
|
|
|
|
|
|
@pytest.mark.parametrize("value", [True, 1.5, "20", -1, 501])
|
|
def test_schema_rejects_coercions_and_unbounded_values(value):
|
|
with pytest.raises(ValidationError):
|
|
routes.DeliveryPolicyUpdate(synchronous_send_max_recipients=value, expected_revision="a" * 64)
|
|
|
|
|
|
def test_route_rolls_back_configuration_and_history_if_audit_fails(policy):
|
|
policy.audit.side_effect = RuntimeError("Audit unavailable")
|
|
with pytest.raises(RuntimeError, match="Audit unavailable"):
|
|
save(policy, 200)
|
|
policy.session.expire_all()
|
|
assert policy.session.get(SystemSettings, "global").settings == {"unrelated": {"enabled": True}}
|
|
|
|
|
|
def test_configuration_catalog_supports_only_known_scope_keys_and_scopes():
|
|
for scope, permission in (("system", "system:settings:write"), ("tenant", "admin:policies:write")):
|
|
key = f"campaign_delivery_policy.{scope}"
|
|
field = classify_configuration_field(key)
|
|
assert field.owner_module == "campaigns" and field.rollback_history_required
|
|
assert plan_configuration_change(key, actor_scopes=(permission,), value={"synchronous_send_max_recipients": 200}).allowed
|
|
assert not plan_configuration_change(key, actor_scopes=(), value={"synchronous_send_max_recipients": 200}).allowed
|
|
assert classify_configuration_field("campaign_delivery_policy.unknown") is None
|
|
|
|
|
|
def test_registered_http_routes_validate_scope_authorization_payload_and_revision(policy):
|
|
app = FastAPI()
|
|
app.include_router(campaign_router, prefix="/api/v1")
|
|
app.dependency_overrides[get_session] = lambda: policy.session
|
|
actor = [policy.admin]
|
|
app.dependency_overrides[get_api_principal] = lambda: actor[0]
|
|
with TestClient(app) as client:
|
|
path = "/api/v1/campaigns/settings/delivery-policy/system"
|
|
response = client.get(path)
|
|
assert response.status_code == 200
|
|
payload = {"synchronous_send_max_recipients": 200, "expected_revision": response.json()["revision"]}
|
|
actor[0] = principal("system:settings:read")
|
|
assert client.put(path, json=payload).status_code == 403
|
|
actor[0] = principal("admin:policies:read", "admin:policies:write")
|
|
assert client.get(path).status_code == 403
|
|
assert client.put(path, json=payload).status_code == 403
|
|
actor[0] = policy.admin
|
|
assert client.put(path, json={**payload, "settings": {"unrelated": "overwrite"}}).status_code == 422
|
|
assert client.put(path, json={**payload, "synchronous_send_max_recipients": True}).status_code == 422
|
|
saved = client.put(path, json=payload)
|
|
assert saved.status_code == 200 and saved.json()["effective_max_recipients"] == 200
|
|
assert client.put(path, json=payload).status_code == 409
|
|
assert client.get("/api/v1/campaigns/settings/delivery-policy/user").status_code == 422
|
|
actor[0] = principal()
|
|
assert client.get(path).status_code == 403
|
|
assert policy.audit.call_args.kwargs["object_type"] == "campaign_delivery_policy"
|