olog/co3
1
0
Fork 0
Code Issues 2 Pull Requests Packages Projects Releases Wiki Activity

add inital test suite, fix minor Mapping group bug

Browse Source
This commit is contained in:
Sam G. 2024-04-18 02:06:42 -07:00
parent 8e4554ed0d
commit 6caf2631b8
17 changed files with 238 additions and 537 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
Makefile
View File

@ -14,3 +14,6 @@ docs-clean:
make -C docs/ clean
## ------------------------------------------ ##
## ----------------- tests ------------------ ##
test:
pytest --pyargs tests -v

13
co3/co3.py
View File

@ -12,22 +12,9 @@ import logging
from collections import defaultdict
from functools import wraps, partial
#from localsys.db.schema import tables
logger = logging.getLogger(__name__)
#def register_format(_format):
# def decorator(func):
# self.collate.format_map[_format] = func
#
# @wraps(func)
# def register(*args, **kwargs):
# return func(*args, **kwargs)
#
# return register
# return decorator
def collate(action_key, action_groups=None):
def decorator(func):
nonlocal action_groups

30
co3/managers/sql.py
View File

@ -35,13 +35,10 @@ Note: Options for insert/update model
build, then single thread bulk INSERT. (**Note**: this is what the method does).
'''
from pathlib import Path
from collections import defaultdict
import time
import logging
import threading
import math
import time
import pprint
from pathlib import Path
from concurrent.futures import wait, as_completed
import sqlalchemy as sa
@ -98,7 +95,11 @@ class SQLManager(RelationalManager[SQLTable]):
def add_router(self, router):
self.routers.append(router)
def recreate(self, schema: Schema[SQLTable], engine: SQLEngine):
def recreate(
self,
schema: Schema[SQLTable],
engine: SQLEngine
) -> None:
'''
Ideally this remains open, as we can't necessarily rely on a SQLAlchemy metadata
object for all kinds of SQLDatabases (would depend on the backend, for instance).
@ -118,16 +119,22 @@ class SQLManager(RelationalManager[SQLTable]):
connection,
component,
inserts: list[dict],
commit=True
):
'''
Parameters:
'''
with self._insert_lock:
connection.execute(
res = connection.execute(
sa.insert(component.obj),
inserts
)
if commit:
connection.commit()
return res
def insert_many(self, connection, inserts: dict):
'''
Perform provided table inserts.
@ -139,9 +146,10 @@ class SQLManager(RelationalManager[SQLTable]):
if total_inserts < 1: return
logger.info(f'Total of {total_inserts} sync inserts to perform')
start = time.time()
# TODO: add some exception handling? may be fine w default propagation
start = time.time()
res_list = []
with self._insert_lock:
for component in inserts:
comp_inserts = inserts[component]
@ -151,10 +159,14 @@ class SQLManager(RelationalManager[SQLTable]):
f'Inserting {len(comp_inserts)} out-of-date entries into component "{component}"'
)
self.insert(connection, component, comp_inserts)
res = self.insert(connection, component, comp_inserts, commit=False)
res_list.append(res)
connection.commit()
logger.info(f'Insert transaction completed successfully in {time.time()-start:.2f}s')
return res_list
def _file_sync_bools(self):
synced_bools = []
fpaths = utils.paths.iter_nested_paths(self.collector.basepath, no_dir=True)

20
co3/mapper.py
View File

@ -131,7 +131,7 @@ class Mapper[C: Component]:
type_list: list[type[CO3]],
attr_name_map: Callable[[type[CO3]], str | C],
coll_name_map: Callable[[type[CO3], str], str | C] | None = None,
):
) -> None:
'''
Auto-register a set of types to the Mapper's attached Schema. Associations are
made from types to both attribute and collation component names, through
@ -194,8 +194,10 @@ class Mapper[C: Component]:
Returns: dict with keys and values relevant for associated SQLite tables
'''
# default is to have no actions
if action_keys is None:
action_keys = list(obj.action_map.keys())
action_keys = []
#action_keys = list(obj.action_registry.keys())
receipts = []
for _cls in reversed(obj.__class__.__mro__[:-2]):
@ -218,7 +220,7 @@ class Mapper[C: Component]:
if collation_data is None:
continue
_, action_groups = obj.action_registry[action_key]
_, action_groups = obj.action_registry.get(action_key, (None, []))
for action_group in action_groups:
collation_component = self.get_collation_comp(_cls, group=action_group)
@ -329,7 +331,7 @@ class ComposableMapper[C: ComposableComponent](Mapper[C]):
def compose(
self,
obj: CO3,
obj: CO3 | type[CO3],
action_groups: list[str] | None = None,
*compose_args,
**compose_kwargs,
@ -344,12 +346,14 @@ class ComposableMapper[C: ComposableComponent](Mapper[C]):
chain (for components / sa.Relationships, it's a little easier).
Parameters:
outer: whether to use outer joins down the chain
conversion: whether to return conversion joins or base primitives
full: whether to return fully connected primitive and conversion table
obj: either a CO3 instance or a type reference
'''
class_ref = obj
if isinstance(obj, CO3):
class_ref = obj.__class__
attr_comp_agg = None
for _cls in reversed(obj.__class__.__mro__[:-2]):
for _cls in reversed(class_ref.__mro__[:-2]):
attr_comp = self.get_attribute_comp(_cls)
# require an attribute component for type consideration

204
examples/.ipynb_checkpoints/database-checkpoint.ipynb
View File

@ -1,204 +0,0 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": 1,
"id": "6f6fbc7e-4fb9-4353-b2ee-9ea819a3c896",
"metadata": {},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"/home/smgr/.pyenv/versions/co4/lib/python3.12/site-packages/tqdm/auto.py:21: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html\n",
" from .autonotebook import tqdm as notebook_tqdm\n"
]
}
],
"source": [
"import vegetables"
]
},
{
"cell_type": "code",
"execution_count": 2,
"id": "88fd0ea8-9c94-4569-a51b-823a04f32f55",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"{'age': 3}"
]
},
"execution_count": 2,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"tomato = vegetables.Tomato('t1', 5)\n",
"\n",
"# test a register collation action\n",
"tomato.collate('ripe')"
]
},
{
"cell_type": "code",
"execution_count": 3,
"id": "348926d9-7137-4eff-a919-508788553dd2",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"['afff61ec-e0b0-44bb-9f0d-06008a82f6a5',\n",
" '5edfa13c-0eb1-4bbc-b55e-1550ff7df3d2',\n",
" '4568a2d4-eb41-4b15-9a29-e3e22906c661']"
]
},
"execution_count": 3,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"vegetables.vegetable_mapper.collect(tomato, ['ripe'])"
]
},
{
"cell_type": "code",
"execution_count": 4,
"id": "4e5e7319-11bf-4051-951b-08c84e9f3874",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"<Component (SQLTable)> vegetable+tomato+tomato_aging_states+tomato_cooking_states"
]
},
"execution_count": 4,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"vegetables.vegetable_mapper.compose(tomato, action_groups=['aging', 'cooking'])"
]
},
{
"cell_type": "code",
"execution_count": 5,
"id": "aa290686-8074-4038-a3cc-ce6817844653",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"[Column('id', Integer(), table=<vegetable>, primary_key=True, nullable=False),\n",
" Column('name', String(), table=<vegetable>),\n",
" Column('color', String(), table=<vegetable>),\n",
" Column('id', Integer(), table=<tomato>, primary_key=True, nullable=False),\n",
" Column('name', String(), ForeignKey('vegetable.name'), table=<tomato>),\n",
" Column('radius', Integer(), table=<tomato>),\n",
" Column('id', Integer(), table=<tomato_aging_states>, primary_key=True, nullable=False),\n",
" Column('name', String(), ForeignKey('tomato.name'), table=<tomato_aging_states>),\n",
" Column('state', String(), table=<tomato_aging_states>),\n",
" Column('age', Integer(), table=<tomato_aging_states>)]"
]
},
"execution_count": 5,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"list(vegetables.vegetable_mapper.compose(tomato, action_groups=['aging']).obj.columns)"
]
},
{
"cell_type": "code",
"execution_count": 6,
"id": "f3c7e37d-ba9e-4bae-ae44-adc922bf5f4c",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"(vegetables.Tomato, vegetables.Vegetable, co3.co3.CO3, object)"
]
},
"execution_count": 6,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"tomato.__class__.__mro__"
]
},
{
"cell_type": "code",
"execution_count": 7,
"id": "c21d2c54-39e2-4de3-93bc-763896ed348e",
"metadata": {},
"outputs": [],
"source": [
"from co3.databases import SQLDatabase\n",
"\n",
"db = SQLDatabase('sqlite://', echo=True)"
]
},
{
"cell_type": "code",
"execution_count": 8,
"id": "a785d202-99d3-4ae7-859e-ee22b481f8df",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"<contextlib._GeneratorContextManager at 0x7dd5c619be60>"
]
},
"execution_count": 8,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"db.recreate(vegetable_schema) "
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "cda01cb0-1666-4cb1-aa64-bcdca871aff5",
"metadata": {},
"outputs": [],
"source": []
}
],
"metadata": {
"kernelspec": {
"display_name": "co3",
"language": "python",
"name": "co3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.12.2"
}
},
"nbformat": 4,
"nbformat_minor": 5
}

145
examples/.ipynb_checkpoints/mapper-checkpoint.ipynb
View File

@ -1,145 +0,0 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": 1,
"id": "e02ccafe-e04d-4312-acba-e41cf7b1c021",
"metadata": {},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"/home/smgr/.pyenv/versions/co4/lib/python3.12/site-packages/tqdm/auto.py:21: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html\n",
" from .autonotebook import tqdm as notebook_tqdm\n"
]
}
],
"source": [
"from co3 import Mapper\n",
"\n",
"import vegetables"
]
},
{
"cell_type": "code",
"execution_count": 2,
"id": "7d80f7b9-7458-4ad4-8c1a-3ea56e796b4e",
"metadata": {},
"outputs": [],
"source": [
"vegetable_mapper = Mapper(\n",
" vegetables.Vegetable,\n",
" vegetables.vegetable_schema\n",
")\n",
"\n",
"vegetable_mapper.attach(\n",
" vegetables.Vegetable,\n",
" vegetables.vegetable_table,\n",
")"
]
},
{
"cell_type": "code",
"execution_count": 4,
"id": "f9408562-bf50-4522-909c-318557f85948",
"metadata": {},
"outputs": [],
"source": [
"# manually attach component\n",
"vegetable_mapper.attach(\n",
" vegetables.Tomato,\n",
" vegetables.tomato_table,\n",
" coll_groups={\n",
" 'aging': vegetables.vegetable_schema.get_component('tomato_aging_states'),\n",
" 'cooking': vegetables.vegetable_schema.get_component('tomato_cooking_states'),\n",
" },\n",
")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "05fdd404-87ee-4187-832f-2305272758ae",
"metadata": {},
"outputs": [],
"source": [
"# attach by name in schema\n",
"vegetable_mapper.attach(\n",
" vegetables.Tomato,\n",
" vegetables.tomato_table,\n",
" coll_groups={\n",
" 'aging': 'tomato_aging_states',\n",
" 'cooking': 'tomato_cooking_states',\n",
" },\n",
")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "e9b6af49-a69d-41cc-beae-1b6f171cd2f5",
"metadata": {},
"outputs": [],
"source": [
"# attach entire type hierarchy w/ type->name map\n",
"vegetable_mapper.attach_hierarchy(\n",
"# this might make more sense during init\n",
" lambda x:x.__name__.lower())\n",
")"
]
},
{
"cell_type": "code",
"execution_count": 9,
"id": "2e4336ab-5b5f-484d-815d-164d4b6f40a0",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"defaultdict(dict,\n",
" {vegetables.Tomato: {'aging': <co3.components.SQLTable at 0x7ece94358aa0>,\n",
" 'cooking': <co3.components.SQLTable at 0x7ece94358ad0>}})"
]
},
"execution_count": 9,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"vegetable_mapper.collation_groups"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "d416f9cd-2cb6-4a6e-bab7-86ac21216b8c",
"metadata": {},
"outputs": [],
"source": []
}
],
"metadata": {
"kernelspec": {
"display_name": "co3",
"language": "python",
"name": "co3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.12.2"
}
},
"nbformat": 4,
"nbformat_minor": 5
}

135
examples/.ipynb_checkpoints/vegetables-checkpoint.py
View File

@ -1,135 +0,0 @@
'''
just remembered tomatos aren't vegetables. whoops
'''
import random
import sqlalchemy as sa
from co3.schemas import SQLSchema
from co3 import CO3, collate, Mapper, ComposableMapper
from co3 import util
class Vegetable(CO3):
def __init__(self, name, color):
self.name = name
self.color = color
class Tomato(Vegetable):
def __init__(self, name, radius):
super().__init__(name, 'red')
self.radius = radius
@property
def attributes(self):
return vars(self)
def collation_attributes(self, action_key, action_grounp):
return {
'name': self.name,
'state': action_key,
}
@collate('ripe', action_groups=['aging'])
def ripen(self):
return {
'age': random.randint(1, 6)
}
@collate('rotten', action_groups=['aging'])
def rot(self):
return {
'age': random.randint(4, 9)
}
@collate('diced', action_groups=['cooking'])
def dice(self):
return {
'pieces': random.randint(2, 12)
}
type_list = [Vegetable, Tomato]
'''
VEGETABLE
|
TOMATO -- AGING
|
-- COOKING
Note: foreign keys need to represent values that could be known by objects _without_ first interacting
with a DB. This is slightly non-standard, given how common it is to depend on another table's integer ID
(typically a value assigned by the DB using an autoincrement, for example, and not specified explicitly
within the insertion body). As a result, SQLTable components need to be able to operate by another unique
key when expected to connect to other tables in the hierarchy. Below we use `name` with a UNIQUE constraint
for this purpose. Note that having an integer `id` is still perfectly okay so that a table can manage
uniqueness of its own rows by default.
'''
metadata = sa.MetaData()
vegetable_table = sa.Table(
'vegetable',
metadata,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('name', sa.String, unique=True),
sa.Column('color', sa.String),
)
tomato_table = sa.Table(
'tomato',
metadata,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('name', sa.String, util.db.deferred_cd_fkey('vegetable.name'), unique=True),
sa.Column('radius', sa.Integer),
)
tomato_aging_table = sa.Table(
'tomato_aging_states',
metadata,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('name', sa.String, util.db.deferred_cd_fkey('tomato.name'), unique=True),
sa.Column('state', sa.String),
sa.Column('age', sa.Integer),
)
tomato_cooking_table = sa.Table(
'tomato_cooking_states',
metadata,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('name', sa.String, util.db.deferred_cd_fkey('tomato.name'), unique=True),
sa.Column('state', sa.String),
sa.Column('pieces', sa.Integer),
)
vegetable_schema = SQLSchema.from_metadata(metadata)
def general_compose_map(c1, c2):
return c1.obj.c.name == c2.obj.c.name
vegetable_mapper = ComposableMapper(
vegetable_schema,
attr_compose_map=general_compose_map,
coll_compose_map=general_compose_map,
)
def attr_name_map(cls):
return f'{cls.__name__.lower()}'
def coll_name_map(cls, action_group):
return f'{cls.__name__.lower()}_{action_group}_states'
vegetable_mapper.attach_many(
type_list,
attr_name_map,
coll_name_map,
)
'''
new mapping type for Mapper attachment:
Callable[ [type[CO3], str|None], tuple[str, tuple[str], tuple[str]]]
tail tuples to associate column names from central table to collation
this should complete the auto-compose horizontally
'''

2
examples/database.ipynb → notebooks/database.ipynb
View File

@ -196,7 +196,7 @@
"source": [
"with db.engine.connect() as connection:\n",
" db.manager.insert_many(\n",
" connection, \n",
" connection,\n",
" vegetables.vegetable_mapper.collector.inserts,\n",
" )"
]

4
examples/mapper.ipynb → notebooks/mapper.ipynb
View File

@ -45,9 +45,7 @@
{
"cell_type": "markdown",
"id": "d2672422-3596-4eab-ac44-5da617f74b80",
"metadata": {
"jp-MarkdownHeadingCollapsed": true
},
"metadata": {},
"source": [
"## Explicit example steps"
]

0
tests/co4_example.py → tests/co4_api_demo.py
View File

19
tests/imports.py
View File

@ -1,19 +0,0 @@
from co3 import Accessor
from co3 import CO3
from co3 import Collector
from co3 import Database
from co3 import Indexer
from co3 import Manager
from co3 import Mapper
from co3 import Relation
from co3.accessors import SQLAccessor
from co3.databases import SQLDatabase
from co3.databases import SQLiteDatabase
from co3.managers import SQLManager
from co3.relations import TabularRelation
from co3.relations import SQLTable

0
examples/vegetables.py → tests/setups/vegetables.py
View File

57
tests/test_co3.py Normal file
View File

@ -0,0 +1,57 @@
from co3.components import Relation
from setups import vegetables as veg
def test_mapper_getters():
veg_comp = veg.vegetable_schema.get_component('vegetable')
tom_comp = veg.vegetable_schema.get_component('tomato')
assert veg.vegetable_mapper.get_attribute_comp(veg.Vegetable) is veg_comp
assert veg.vegetable_mapper.get_attribute_comp(veg.Tomato) is tom_comp
tom_aging = veg.vegetable_schema.get_component('tomato_aging_states')
tom_cooking = veg.vegetable_schema.get_component('tomato_cooking_states')
assert veg.vegetable_mapper.get_collation_comp(veg.Tomato, 'aging') is tom_aging
assert veg.vegetable_mapper.get_collation_comp(veg.Tomato, 'cooking') is tom_cooking
def test_mapper_attach():
assert veg.vegetable_mapper.attach(
veg.Tomato,
'tomato',
coll_groups={
'aging': 'tomato_aging_states',
'cooking': 'tomato_cooking_states',
},
) is None
def test_mapper_attach_many():
assert veg.vegetable_mapper.attach_many(
[veg.Vegetable, veg.Tomato],
lambda t: f'{t.__name__.lower()}'
) is None
def test_mapper_collect():
tomato = veg.Tomato('t1', 10)
receipts = veg.vegetable_mapper.collect(tomato)
assert len(receipts) == 2
# attempt to retrieve receipts one at a time
res1 = veg.vegetable_mapper.collector.collect_inserts([receipts[0]])
assert len(res1) == 1 # should be just one match
assert len(res1[next(iter(res1.keys()))]) == 1 # and one dict for matching comp
# try again, check no persistent match
res1 = veg.vegetable_mapper.collector.collect_inserts([receipts[0]])
assert len(res1) == 0 # should be no matches for the same receipt
res2 = veg.vegetable_mapper.collector.collect_inserts([receipts[1]])
assert len(res2) == 1
assert len(res2[next(iter(res2.keys()))]) == 1

36
tests/test_database.py Normal file
View File

@ -0,0 +1,36 @@
from co3.components import Relation
from co3.databases import SQLDatabase
from setups import vegetables as veg
db = None
def test_database_init():
global db
db = SQLDatabase('sqlite://')
assert True
def test_database_recreate():
db.recreate(veg.vegetable_schema)
assert True
def test_database_insert():
tomato = veg.Tomato('t1', 5)
veg.vegetable_mapper.collect(tomato)
with db.engine.connect() as connection:
assert db.manager.insert_many(
connection,
veg.vegetable_mapper.collector.inserts,
) is not None
def test_database_access():
agg_table = veg.vegetable_mapper.compose(veg.Tomato)
with db.engine.connect() as connection:
assert db.accessor.select(
connection,
agg_table,
) is not None

25
tests/test_imports.py Normal file
View File

@ -0,0 +1,25 @@
def test_import():
from co3 import Accessor
from co3 import CO3
from co3 import Collector
from co3 import Database
from co3 import Indexer
from co3 import Manager
from co3 import Mapper
from co3 import Component
from co3.accessors import SQLAccessor
from co3.accessors import FTSAccessor
from co3.accessors import VSSAccessor
from co3.databases import SQLDatabase
from co3.databases import SQLiteDatabase
from co3.managers import SQLManager
from co3.components import ComposableComponent
from co3.components import Relation
from co3.components import SQLTable
assert True

56
tests/test_mapper.py Normal file
View File

@ -0,0 +1,56 @@
from co3.components import Relation
from setups import vegetables as veg
def test_mapper_getters():
veg_comp = veg.vegetable_schema.get_component('vegetable')
tom_comp = veg.vegetable_schema.get_component('tomato')
assert veg.vegetable_mapper.get_attribute_comp(veg.Vegetable) is veg_comp
assert veg.vegetable_mapper.get_attribute_comp(veg.Tomato) is tom_comp
tom_aging = veg.vegetable_schema.get_component('tomato_aging_states')
tom_cooking = veg.vegetable_schema.get_component('tomato_cooking_states')
assert veg.vegetable_mapper.get_collation_comp(veg.Tomato, 'aging') is tom_aging
assert veg.vegetable_mapper.get_collation_comp(veg.Tomato, 'cooking') is tom_cooking
def test_mapper_attach():
assert veg.vegetable_mapper.attach(
veg.Tomato,
'tomato',
coll_groups={
'aging': 'tomato_aging_states',
'cooking': 'tomato_cooking_states',
},
) is None
def test_mapper_attach_many():
assert veg.vegetable_mapper.attach_many(
[veg.Vegetable, veg.Tomato],
lambda t: f'{t.__name__.lower()}'
) is None
def test_mapper_collect():
tomato = veg.Tomato('t1', 10)
receipts = veg.vegetable_mapper.collect(tomato)
assert len(receipts) == 2
# attempt to retrieve receipts one at a time
res1 = veg.vegetable_mapper.collector.collect_inserts([receipts[0]])
assert len(res1) == 1 # should be just one match
assert len(res1[next(iter(res1.keys()))]) == 1 # and one dict for matching comp
# try again, check no persistent match
res1 = veg.vegetable_mapper.collector.collect_inserts([receipts[0]])
assert len(res1) == 0 # should be no matches for the same receipt
res2 = veg.vegetable_mapper.collector.collect_inserts([receipts[1]])
assert len(res2) == 1
assert len(res2[next(iter(res2.keys()))]) == 1

26
tests/test_schema.py Normal file
View File

@ -0,0 +1,26 @@
from co3.components import Relation
from setups import vegetables as veg
def test_schema_get():
veg_comp_raw = veg.vegetable_schema._component_map.get('vegetable')
veg_comp = veg.vegetable_schema.get_component('vegetable')
assert veg_comp_raw is veg_comp
def test_schema_contains():
vegetable_comp = veg.vegetable_schema.get_component('vegetable')
tomato_comp = veg.vegetable_schema.get_component('tomato')
tomato_aging_comp = veg.vegetable_schema.get_component('tomato_aging_states')
assert vegetable_comp in veg.vegetable_schema
assert tomato_comp in veg.vegetable_schema
assert tomato_aging_comp in veg.vegetable_schema
def test_schema_add():
veg.vegetable_schema.add_component(Relation[int]('a', 1))
veg.vegetable_schema.add_component(Relation[int]('b', 2))
assert veg.vegetable_schema.get_component('a') is not None
assert veg.vegetable_schema.get_component('b') is not None