| from decimal import Decimal |
| import time |
| import pytest |
| from cortex_ai.orchestration import * |
| from cortex_ai.orchestration.errors import PolicyDenied |
|
|
| class EchoProvider(Provider): |
| id = "echo" |
| capabilities = frozenset({Capability.VISION, Capability.VIDEO_GENERATION}) |
| local = True |
| def health(self): return True |
| def generate(self, request): return ProviderResult({"echo": request["text"]}, self.id, {"test": True}) |
|
|
| def test_router_detects_mixed_request_and_privacy(): |
| req = CapabilityRouter().route("Analyse cette facture confidentielle et génère une vidéo") |
| assert Capability.DOCUMENT_ANALYSIS in req.capabilities |
| assert Capability.VIDEO_GENERATION in req.capabilities |
| assert req.requires_async_job is True |
| assert req.requires_async is True |
| assert req.privacy_mode == PrivacyMode.LOCAL_ONLY |
|
|
| def test_registry_selects_provider_and_engine_returns_job_for_long_task(): |
| jobs = JobManager() |
| engine = OrchestrationEngine(CapabilityRouter(), ProviderRegistry([EchoProvider()]), Policy(), jobs) |
| job = engine.execute("generate a video", preferred_provider="echo") |
| assert job.provider == "echo" |
| for _ in range(20): |
| if jobs.get(job.id).status == JobStatus.COMPLETED: break |
| time.sleep(0.01) |
| assert jobs.get(job.id).result.output["echo"] == "generate a video" |
| jobs.shutdown() |
|
|
| def test_local_only_policy_denies_external_provider(): |
| class External(EchoProvider): |
| id = "external"; local = False |
| with pytest.raises(PolicyDenied): |
| OrchestrationEngine(CapabilityRouter(), ProviderRegistry([External()]), Policy(PrivacyMode.LOCAL_ONLY), JobManager()).execute("describe this image", preferred_provider="external") |
|
|
| def test_money_is_decimal_and_supports_gnf(): |
| total = add(Money(Decimal("10.005"), "USD"), Money(Decimal("2.00"), "USD")) |
| assert total == Money(Decimal("12.01"), "USD") |
| assert convert(Money(Decimal("100"), "USD"), "GNF", Decimal("8500")).amount == Decimal("850000.00") |
|
|
