333 lines
8.1 KiB
Python
333 lines
8.1 KiB
Python
import sys
|
|
import os
|
|
import pytest
|
|
|
|
sys.path.insert(0, os.path.realpath(os.path.dirname(__file__))+'/../../')
|
|
|
|
#from settings import config
|
|
from cuchulu.libraries.db.webmodel import WebModel
|
|
from cuchulu.libraries.db import corefields
|
|
|
|
|
|
class ExampleModel(WebModel):
|
|
|
|
def __init__(self, connection):
|
|
|
|
super().__init__(connection)
|
|
|
|
# I can change other fields here, how the name.
|
|
|
|
self.register(corefields.CharField('title'))
|
|
self.register(corefields.CharField('content'))
|
|
|
|
class ForeignKeyExampleModel(WebModel):
|
|
|
|
def __init__(self, connection):
|
|
|
|
super().__init__(connection)
|
|
|
|
# I can change other fields here, how the name.
|
|
|
|
self.register(corefields.CharField('name'))
|
|
self.register(corefields.ForeignKeyField('example_id', ExampleModel(connection), size=11, required=False, identifier_field='id', named_field="id", select_fields=[]))
|
|
|
|
|
|
class ExampleModel2(WebModel):
|
|
|
|
def __init__(self, connection):
|
|
|
|
super().__init__(connection)
|
|
|
|
# I can change other fields here, how the name.
|
|
|
|
self.register(corefields.CharField('title'))
|
|
self.register(corefields.CharField('content'))
|
|
|
|
# Define fixture for get data for mysql connection via command line
|
|
|
|
# WebModel.connections={'default': {'name': 'default', 'host': 'localhost', 'user': 'root', 'password': '', 'db': 'cuchuludev_db', 'charset': 'utf8mb4', 'set_connection': False, 'db_type': 'pymysql'} }
|
|
|
|
def test_test_table(webmodel_conn):
|
|
|
|
print('Test table')
|
|
|
|
model=ExampleModel(webmodel_conn)
|
|
|
|
assert prepare_table(model)
|
|
|
|
# Delete tables
|
|
|
|
|
|
def test_test_insert(webmodel_conn):
|
|
|
|
print('Test Insert')
|
|
|
|
model=ExampleModel(webmodel_conn)
|
|
|
|
assert prepare_table(model)
|
|
#model.query('use pruebas_db')
|
|
|
|
assert insert_row(model)
|
|
|
|
def test_test_insert_id(webmodel_conn):
|
|
|
|
model=ExampleModel(webmodel_conn)
|
|
|
|
assert prepare_table(model)
|
|
|
|
#model.query('use pruebas_db')
|
|
|
|
assert insert_row(model)
|
|
|
|
assert model.insert_id()==1
|
|
|
|
def test_test_update(webmodel_conn):
|
|
|
|
model=ExampleModel(webmodel_conn)
|
|
|
|
assert prepare_table(model)
|
|
|
|
assert insert_row(model)
|
|
|
|
post_update={'title': 'Example title Updated', 'content': 'New content Updated'}
|
|
|
|
model.conditions=['WHERE id=%s', [1]]
|
|
|
|
assert model.update(post_update)
|
|
|
|
def test_test_count(webmodel_conn):
|
|
|
|
model=ExampleModel(webmodel_conn)
|
|
|
|
assert prepare_table(model)
|
|
|
|
assert insert_row(model)
|
|
|
|
assert model.select_count()==1
|
|
|
|
def test_test_select_a_row(webmodel_conn):
|
|
|
|
model=ExampleModel(webmodel_conn)
|
|
|
|
assert prepare_table(model)
|
|
|
|
assert insert_row(model)
|
|
|
|
assert model.select_count()==1
|
|
|
|
assert model.select_a_row(1, ['title', 'inexistent_field'])=={'title': 'Example title'}
|
|
|
|
def test_test_select_a_row_where(webmodel_conn):
|
|
|
|
model=ExampleModel(webmodel_conn)
|
|
|
|
assert prepare_table(model)
|
|
|
|
assert insert_row(model)
|
|
|
|
assert model.set_conditions('WHERE id=%s', [1]).select_a_row_where(['title'])=={'title': 'Example title'}
|
|
|
|
def test_test_select_to_array(webmodel_conn):
|
|
|
|
model=ExampleModel(webmodel_conn)
|
|
|
|
assert prepare_table(model)
|
|
|
|
assert insert_row(model)
|
|
|
|
assert model.select_to_array(['title', 'content'])==[{'id': 1, 'title': 'Example title', 'content': 'New content'}]
|
|
|
|
def test_test_reset_conditions(webmodel_conn):
|
|
|
|
model=ExampleModel(webmodel_conn)
|
|
|
|
assert prepare_table(model)
|
|
|
|
assert insert_row(model)
|
|
|
|
model.set_conditions('WHERE id=%s', [1])
|
|
|
|
model.yes_reset_conditions=True
|
|
|
|
model.reset_conditions()
|
|
|
|
assert model.conditions==['WHERE 1=1', []]
|
|
|
|
def test_test_simple_select(webmodel_conn):
|
|
|
|
model=ExampleModel(webmodel_conn)
|
|
|
|
assert prepare_table(model)
|
|
|
|
assert insert_row(model)
|
|
|
|
with model.select() as cur:
|
|
|
|
row=model.fetch(cur)
|
|
|
|
assert row=={'id': 1, 'title': 'Example title', 'content': 'New content'}
|
|
|
|
def test_test_element_exists(webmodel_conn):
|
|
|
|
model=ExampleModel(webmodel_conn)
|
|
|
|
assert prepare_table(model)
|
|
|
|
assert insert_row(model)
|
|
|
|
assert model.element_exists(1)
|
|
|
|
def test_test_delete(webmodel_conn):
|
|
|
|
model=ExampleModel(webmodel_conn)
|
|
|
|
assert prepare_table(model)
|
|
|
|
assert insert_row(model)
|
|
|
|
model.conditions=['WHERE id=%s', [2]]
|
|
|
|
assert model.delete()==False
|
|
|
|
model.conditions=['WHERE id=%s', [1]]
|
|
|
|
assert model.delete()==True
|
|
|
|
def test_test_drop(webmodel_conn):
|
|
|
|
model=ExampleModel(webmodel_conn)
|
|
|
|
assert prepare_table(model)
|
|
|
|
assert insert_row(model)
|
|
|
|
assert model.drop()
|
|
|
|
def test_test_update_table(webmodel_conn):
|
|
|
|
model=ExampleModel(webmodel_conn)
|
|
|
|
assert prepare_table(model)
|
|
|
|
assert insert_row(model)
|
|
|
|
fields_to_modify=[]
|
|
fields_to_add_index=[]
|
|
fields_to_add_constraint=[]
|
|
fields_to_add_unique=[]
|
|
fields_to_delete_index=[]
|
|
fields_to_delete_unique=[]
|
|
fields_to_delete_constraint=[]
|
|
fields_to_delete=[]
|
|
|
|
model.register(corefields.IntegerField('description'))
|
|
|
|
model.update_table(['description'], fields_to_modify, fields_to_add_index, fields_to_add_constraint, fields_to_add_unique, fields_to_delete_index, fields_to_delete_unique, fields_to_delete_constraint, fields_to_delete)
|
|
|
|
model.register(corefields.CharField('description'))
|
|
|
|
model.update_table([], ['description'], ['description'], [], ['description'], fields_to_delete_index, fields_to_delete_unique, fields_to_delete_constraint, fields_to_delete)
|
|
|
|
model.update_table([], fields_to_modify, fields_to_add_index, fields_to_add_constraint, fields_to_add_unique, ['description'], ['description'], fields_to_delete_constraint, ['description'])
|
|
|
|
assert 1
|
|
|
|
def test_test_set_conditions(webmodel_conn):
|
|
|
|
model=ExampleModel(webmodel_conn)
|
|
|
|
assert prepare_table(model)
|
|
|
|
assert insert_row(model)
|
|
|
|
cur=model.set_conditions('where id=%s', [4]).select()
|
|
|
|
assert cur
|
|
|
|
cur.close()
|
|
|
|
assert model.drop()
|
|
|
|
def test_test_foreignkeys(webmodel_conn):
|
|
|
|
connection=webmodel_conn
|
|
|
|
model=ExampleModel(connection)
|
|
foreignkey=ForeignKeyExampleModel(connection)
|
|
|
|
sql=model.create_table()
|
|
sqlf=foreignkey.create_table()
|
|
|
|
assert model.query(sql)
|
|
assert foreignkey.query(sqlf)
|
|
|
|
for k_field, index in WebModel.arr_sql_index['foreignkeyexamplemodel'].items():
|
|
print("---Added index to "+k_field)
|
|
foreignkey.query(index)
|
|
|
|
for k_set, index_set in WebModel.arr_sql_set_index['foreignkeyexamplemodel'].items():
|
|
|
|
if index_set!="":
|
|
connection.query(index_set)
|
|
print("---Added constraint to "+k_set)
|
|
|
|
model.create_forms()
|
|
|
|
assert model.insert({'title': 'Foreign title', 'content': 'Foreign content'})
|
|
|
|
my_id=model.insert_id()
|
|
|
|
foreignkey.create_forms()
|
|
|
|
assert foreignkey.insert({'example_id': my_id, 'name': 'Relationship'})
|
|
|
|
foreignkey.set_conditions('where example_id=%s', [my_id])
|
|
|
|
assert foreignkey.select_count()==1
|
|
|
|
model.set_conditions('where id=%s', [my_id])
|
|
|
|
assert model.delete()
|
|
|
|
foreignkey.set_conditions('where example_id=%s', [my_id])
|
|
|
|
assert foreignkey.select_count()==0
|
|
|
|
assert foreignkey.drop()
|
|
assert model.drop()
|
|
|
|
def test_test_check_connections(webmodel_conn):
|
|
|
|
connection=webmodel_conn
|
|
|
|
model=ExampleModel(connection)
|
|
|
|
model2=ExampleModel2(connection)
|
|
|
|
sql=model.create_table()
|
|
sql2=model2.create_table()
|
|
|
|
assert model.query(sql)
|
|
assert model2.query(sql2)
|
|
|
|
assert model.drop()
|
|
assert model2.drop()
|
|
|
|
print('last_test')
|
|
|
|
@pytest.mark.skip
|
|
def prepare_table(model):
|
|
|
|
sql=model.create_table()
|
|
|
|
return model.query(sql)
|
|
|
|
@pytest.mark.skip
|
|
def insert_row(model):
|
|
|
|
post={'title': 'Example title', 'content': 'New content'}
|
|
|
|
model.set_valid_fields()
|
|
|
|
return model.insert(post)
|
|
|