mi
95 lines
3.4 KiB
Python
95 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from app.config import Settings
|
|
|
|
|
|
def test_disabled_mode_needs_no_secrets() -> None:
|
|
settings = Settings(enabled=False, mode="disabled")
|
|
assert settings.enabled is False
|
|
|
|
|
|
def test_full_mode_rejects_portal_host_mismatch() -> None:
|
|
with pytest.raises(ValidationError, match="approved portal host"):
|
|
Settings(
|
|
enabled=True,
|
|
mode="full",
|
|
database_url="postgresql+asyncpg://u:p@db/han",
|
|
crm_rest_webhook_url="https://evil.example/rest/1/token/",
|
|
contact_receiver_token="contact",
|
|
alert_receiver_token="alert",
|
|
service_token="service",
|
|
portal_host="portal.example",
|
|
portal_member_id="member_12345678",
|
|
public_base_url="https://sync.example",
|
|
contact_user_id_field="UF_CRM_1",
|
|
contact_registered_field="UF_CRM_2",
|
|
contact_citizenship_field="UF_CRM_3",
|
|
webhook_allowed_cidrs="203.0.113.0/24",
|
|
)
|
|
|
|
|
|
def test_rest_field_conversion_is_deterministic() -> None:
|
|
assert Settings.rest_field_name("UF_CRM_1778692456") == "ufCrm_1778692456"
|
|
assert Settings.rest_field_name("UF_CRM_6a70c275346a7") == "ufCrm_6a70c275346a7"
|
|
assert Settings.rest_field_name("UF_CRM_AbC123") == "ufCrm_AbC123"
|
|
with pytest.raises(ValueError):
|
|
Settings.rest_field_name("uf_crm_1")
|
|
with pytest.raises(ValueError):
|
|
Settings.rest_field_name("UF_CRM_123_abc")
|
|
|
|
|
|
def test_image_default_allows_compose_process_role_override() -> None:
|
|
service_root = Path(__file__).parents[1]
|
|
dockerfile = (service_root / "Dockerfile").read_text(encoding="utf-8")
|
|
compose = (service_root / "compose.fragment.yaml").read_text(encoding="utf-8")
|
|
production_compose = (service_root.parent / "docker-compose.yml").read_text(encoding="utf-8")
|
|
|
|
assert 'CMD ["han-bitrix-sync-api"]' in dockerfile
|
|
assert 'ENTRYPOINT ["han-bitrix-sync-api"]' not in dockerfile
|
|
for source in (compose, production_compose):
|
|
assert 'command: ["han-bitrix-sync-worker"]' in source
|
|
assert 'command: ["han-bitrix-sync-reconciliation"]' in source
|
|
|
|
|
|
def test_alembic_chain_preserves_legacy_baseline() -> None:
|
|
versions = Path(__file__).parents[1] / "alembic" / "versions"
|
|
revisions: dict[str, str | None] = {}
|
|
|
|
for migration in versions.glob("*.py"):
|
|
assignments = {
|
|
node.target.id: node.value.value
|
|
for node in ast.parse(migration.read_text(encoding="utf-8")).body
|
|
if isinstance(node, ast.AnnAssign)
|
|
and isinstance(node.target, ast.Name)
|
|
and node.target.id in {"revision", "down_revision"}
|
|
and isinstance(node.value, ast.Constant)
|
|
}
|
|
revisions[assignments["revision"]] = assignments["down_revision"]
|
|
|
|
assert revisions == {
|
|
"0001_sync_baseline": None,
|
|
"0001_bitrix_sync_full": "0001_sync_baseline",
|
|
"0002_app_queue_contract": "0001_bitrix_sync_full",
|
|
}
|
|
|
|
|
|
def test_app_queue_migration_does_not_revoke_foreign_schema_privileges() -> None:
|
|
migration = (
|
|
Path(__file__).parents[1]
|
|
/ "alembic"
|
|
/ "versions"
|
|
/ "0002_app_queue_contract.py"
|
|
)
|
|
source = migration.read_text(encoding="utf-8")
|
|
|
|
assert "FROM han_app.entity_external_mapping" in source
|
|
assert "REVOKE" not in source
|
|
assert re.search(r'"[^"]+"\s*:', source) is None
|