Spaces:
Running
Running
File size: 2,235 Bytes
c8365f5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | import os
import sys
import types
import unittest
from unittest.mock import patch
from models.ai_client import AIClient
class _LegacySchemaError(Exception):
pass
class _Result:
def __init__(self, data):
self.data = data
class _LegacyQuery:
def __init__(self, rows):
self.rows = rows
self.selects = []
def select(self, columns):
self.selects.append(columns)
if "default_model" in columns:
raise _LegacySchemaError("column ai_providers.default_model does not exist")
return self
def eq(self, *_args, **_kwargs):
return self
def order(self, *_args, **_kwargs):
return self
def execute(self):
return _Result(self.rows)
class _LegacySupabase:
def __init__(self, rows):
self.query = _LegacyQuery(rows)
def table(self, name):
assert name == "ai_providers"
return self.query
class LegacySchemaCompatibilityTests(unittest.TestCase):
def test_detects_only_known_legacy_missing_columns(self):
self.assertTrue(
AIClient._is_legacy_schema_error(
_LegacySchemaError("column ai_providers.default_model does not exist")
)
)
self.assertFalse(
AIClient._is_legacy_schema_error(
_LegacySchemaError("column ai_providers.api_key does not exist")
)
)
def test_legacy_schema_returns_empty_so_current_environment_fallback_stays_authoritative(self):
legacy_supabase = _LegacySupabase([])
fake_supabase = types.SimpleNamespace(
create_client=lambda _url, _key: legacy_supabase,
)
client = AIClient.__new__(AIClient)
with patch.dict(
os.environ,
{"SUPABASE_URL": "https://example.supabase.co", "SUPABASE_SERVICE_ROLE_KEY": "test"},
clear=True,
), patch.dict(sys.modules, {"supabase": fake_supabase}):
providers = client._try_load_from_supabase()
self.assertEqual(legacy_supabase.query.selects, [
"id,name,api_key,base_url,default_model,tier,purpose",
])
self.assertEqual(providers, [])
if __name__ == "__main__":
unittest.main()
|