mirror of
https://github.com/hanzoai/datastore-orm.git
synced 2026-09-30 21:40:30 +00:00
zeekay
dda3979851
This is our fork, so it should read like it. `import datastore_orm` instead of `from infi.clickhouse_orm import ...`: `infi` was the original vendor's org namespace, and carrying someone else's org name through every import in the consuming codebase is not a fork, it is a vendored copy that never got renamed. The `infi` namespace package is gone entirely (src/infi/__init__.py with it), the distribution is `datastore-orm`, and the metadata points at Hanzo rather than at the upstream repo and author. NOT renamed, deliberately: the migration history tables are still `infi_clickhouse_orm_migrations`. That name is persisted state — 448 rows in production key off it — so changing it is a data migration, not a rename. Doing it carelessly reproduces the exact failure this fork just hit, where a package rename orphaned the history and silently blocked every migration.
78 lines
2.7 KiB
Python
78 lines
2.7 KiB
Python
import unittest
|
|
from datetime import date
|
|
|
|
from datastore_orm.database import Database
|
|
from datastore_orm.models import Model
|
|
from datastore_orm.fields import *
|
|
from datastore_orm.engines import *
|
|
|
|
|
|
class ArrayFieldsTest(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.database = Database('test-db', log_statements=True)
|
|
self.database.create_table(ModelWithArrays)
|
|
|
|
def tearDown(self):
|
|
self.database.drop_database()
|
|
|
|
def test_insert_and_select(self):
|
|
instance = ModelWithArrays(
|
|
date_field='2016-08-30',
|
|
arr_str=['goodbye,', 'cruel', 'world', 'special chars: ,"\\\'` \n\t\\[]'],
|
|
arr_date=['2010-01-01'],
|
|
)
|
|
self.database.insert([instance])
|
|
query = 'SELECT * from $db.modelwitharrays ORDER BY date_field'
|
|
for model_cls in (ModelWithArrays, None):
|
|
results = list(self.database.select(query, model_cls))
|
|
self.assertEqual(len(results), 1)
|
|
self.assertEqual(results[0].arr_str, instance.arr_str)
|
|
self.assertEqual(results[0].arr_int, instance.arr_int)
|
|
self.assertEqual(results[0].arr_date, instance.arr_date)
|
|
|
|
def test_conversion(self):
|
|
instance = ModelWithArrays(
|
|
arr_int=('1', '2', '3'),
|
|
arr_date=['2010-01-01']
|
|
)
|
|
self.assertEqual(instance.arr_str, [])
|
|
self.assertEqual(instance.arr_int, [1, 2, 3])
|
|
self.assertEqual(instance.arr_date, [date(2010, 1, 1)])
|
|
|
|
def test_assignment_error(self):
|
|
instance = ModelWithArrays()
|
|
for value in (7, 'x', [date.today()], ['aaa'], [None]):
|
|
with self.assertRaises(ValueError):
|
|
instance.arr_int = value
|
|
|
|
def test_parse_array(self):
|
|
from datastore_orm.utils import parse_array, unescape
|
|
self.assertEqual(parse_array("[]"), [])
|
|
self.assertEqual(parse_array("[1, 2, 395, -44]"), ["1", "2", "395", "-44"])
|
|
self.assertEqual(parse_array("['big','mouse','','!']"), ["big", "mouse", "", "!"])
|
|
self.assertEqual(parse_array(unescape("['\\r\\n\\0\\t\\b']")), ["\r\n\0\t\b"])
|
|
for s in ("",
|
|
"[",
|
|
"]",
|
|
"[1, 2",
|
|
"3, 4]",
|
|
"['aaa', 'aaa]"):
|
|
with self.assertRaises(ValueError):
|
|
parse_array(s)
|
|
|
|
def test_invalid_inner_field(self):
|
|
for x in (DateField, None, "", ArrayField(Int32Field())):
|
|
with self.assertRaises(AssertionError):
|
|
ArrayField(x)
|
|
|
|
|
|
class ModelWithArrays(Model):
|
|
|
|
date_field = DateField()
|
|
arr_str = ArrayField(StringField())
|
|
arr_int = ArrayField(Int32Field())
|
|
arr_date = ArrayField(DateField())
|
|
|
|
engine = MergeTree('date_field', ('date_field',))
|