diff --git a/paramecio2/tests/conftest.py b/paramecio2/tests/conftest.py new file mode 100644 index 0000000..db324b6 --- /dev/null +++ b/paramecio2/tests/conftest.py @@ -0,0 +1,68 @@ +import pytest +from paramecio2.libraries.db.webmodel import WebModel +from paramecio2.libraries.db import corefields + + +def pytest_addoption(parser): + parser.addoption("--mysql_host", action="store", default="host", help="MySQL host: default localhost") + parser.addoption("--mysql_user", action="store", default="root", help="Mysql User for make the test: default root") + parser.addoption("--mysql_password", action="store", default="", help="Mysql password for make the test: default empty value") + parser.addoption("--mysql_db", action="store", default="test_paramecio_db", help="Mysql Database for execute the test: default test_paramecio_db") + parser.addoption("--mysql_type", action="store", default="pymysql", help="MySQL Python library used: options pymysql mysqldb") + +@pytest.fixture +def webmodel_conn(request): + options={'mysql_host': '', 'mysql_user': '', 'mysql_password': '', 'mysql_db': '', 'mysql_type': ''} + + for k in options: + options[k]=request.config.getoption(k) + + # Get connection + #'db': options['mysql_db'], + + WebModel.connections={'default': {'name': 'default', 'host': options['mysql_host'], 'user': options['mysql_user'], 'password': options['mysql_password'], 'db': options['mysql_db'], 'charset': 'utf8mb4', 'set_connection': False, 'db_type': options['mysql_type']} } + + #conn=WebModel.connection() + + # Create database using raw mysql methods + + if options['mysql_type']=='pymysql': + + import pymysql.cursors + pymysql.install_as_MySQLdb + + conn=pymysql.connect(options['mysql_host'], + user=options['mysql_user'], + passwd=options['mysql_password'], + charset='utf8mb4', + cursorclass=pymysql.cursors.DictCursor) + else: + import MySQLdb.cursors + + conn=MySQLdb.connect(options['mysql_host'], + user=options['mysql_user'], + passwd=options['mysql_password'], + charset='utf8mb4', + cursorclass=MySQLdb.cursors.DictCursor) + + conn.query( + '''CREATE DATABASE {name} + DEFAULT CHARACTER SET {charset}''' + .format( + name=options['mysql_db'], charset='utf8mb4' + ) + ) + + #conn.close() + + def drop_database(): + print('Finish him') + conn.query('DROP DATABASE IF EXISTS %s' % options['mysql_db']) + conn.close() + + request.addfinalizer(drop_database) + + final_conn=WebModel.connection() + + return final_conn + diff --git a/paramecio2/tests/webmodel_test.py b/paramecio2/tests/webmodel_test.py new file mode 100644 index 0000000..1c7c363 --- /dev/null +++ b/paramecio2/tests/webmodel_test.py @@ -0,0 +1,327 @@ +import sys +import os +import pytest + +sys.path.insert(0, os.path.realpath(os.path.dirname(__file__))+'/../../') + +#from settings import config +from paramecio2.libraries.db.webmodel import WebModel +from paramecio2.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': 'sirena', 'db': 'cuchuludev_db', 'charset': 'utf8mb4', 'set_connection': False, 'db_type': 'pymysql'} } + +def test_test_table(webmodel_conn): + + model=ExampleModel(webmodel_conn) + + assert prepare_table(model) + + # Delete tables + + +def test_test_insert(webmodel_conn): + + 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() + +@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) +