mirror of
https://18126008609:longquanjian123@gitee.com/feigong123/aurask.git
synced 2026-04-19 23:28:25 +00:00
47 lines
1.8 KiB
Python
47 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "api"))
|
|
|
|
from aurask.bridge_status import bridge_status
|
|
from aurask.bridges.config import BridgeConfig
|
|
from aurask.bridges.pgvector import PGVectorBridge
|
|
from aurask.bridges.postgres import PostgresBridge
|
|
from aurask.bridges.redis_bridge import RedisBridge
|
|
|
|
|
|
class BridgeContractTests(unittest.TestCase):
|
|
def test_bridge_status_reports_all_required_components(self) -> None:
|
|
status = bridge_status()
|
|
self.assertIn("postgres", status)
|
|
self.assertIn("pgvector", status)
|
|
self.assertIn("redis", status)
|
|
self.assertIn("anythingllm", status)
|
|
self.assertIn("langflow", status)
|
|
|
|
def test_pgvector_contract_requires_tenant_workspace_filters(self) -> None:
|
|
bridge = PGVectorBridge(BridgeConfig.from_env().pgvector)
|
|
contract = bridge.search_contract("tenant_1", "workspace_1")
|
|
self.assertEqual(contract["filters"]["tenant_id"], "tenant_1")
|
|
self.assertEqual(contract["filters"]["workspace_id"], "workspace_1")
|
|
self.assertIn("query_embedding", contract["required"])
|
|
|
|
def test_redis_keys_are_tenant_scoped(self) -> None:
|
|
bridge = RedisBridge(BridgeConfig.from_env().redis)
|
|
key = bridge.idempotency_key("tenant_1", "payment", "tx_hash")
|
|
self.assertIn("tenant_1", key)
|
|
self.assertIn("payment", key)
|
|
|
|
def test_postgres_schema_includes_vector_extension(self) -> None:
|
|
bridge = PostgresBridge(BridgeConfig.from_env().postgres)
|
|
schema = bridge.migration_plan()["schema_sql"]
|
|
self.assertIn("CREATE EXTENSION IF NOT EXISTS vector", schema)
|
|
self.assertIn("quota_ledger", schema)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|