Fixes in tests

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="localhost", 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(host=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
+    

tests/fields_test.py

@@ -0,0 +1,165 @@
+import sys
+import os
+import pytest
+import json
+
+sys.path.insert(0, os.path.realpath(os.path.dirname(__file__))+'/../../')
+
+#from settings import config
+from paramecio2.libraries.db.webmodel import PhangoField, WebModel
+from paramecio2.libraries.db import corefields
+from paramecio2.libraries.db.extrafields.arrayfield import ArrayField
+from paramecio2.libraries.db.extrafields.colorfield import ColorField
+from paramecio2.libraries.db.extrafields.datefield import DateField
+from paramecio2.libraries.db.extrafields.datetimefield import DateTimeField
+from paramecio2.libraries.db.extrafields.dictfield import DictField
+from paramecio2.libraries.db.extrafields.emailfield import EmailField
+from paramecio2.libraries.db.extrafields.i18nfield import I18nField
+from paramecio2.libraries import datetime
+
+class ExampleModel(WebModel):
+    
+    def __init__(self, connection=None):
+
+        super().__init__(connection)
+
+        # I can change other fields here, how the name.
+
+        self.register(corefields.CharField('title'))
+        self.register(corefields.CharField('content'))
+
+
+def test_test_phangofield():
+    
+    field=PhangoField('default', 255)
+
+    assert field.check(' "<trial>" ')=='&quot;&lt;trial&gt;&quot;'
+
+def test_test_integerfield():
+    
+    field=corefields.IntegerField('int')
+    
+    assert field.check("25")=='25'
+    
+    assert field.check("25j")=='0'
+    
+    field.check("25j")
+    
+    assert field.error==True
+
+def test_test_floadfield():
+    
+    field=corefields.FloatField('int')
+    
+    assert field.check('1,5')=='1.5'
+    
+    assert field.check('1.5456')=='1.5456'
+    
+    assert field.check('1.5456tet')=='0'
+
+def test_test_htmlfield():
+    
+    field=corefields.HTMLField('html')
+    
+    assert field.check('<p>"trial"</p><script></script>')=='&lt;p&gt;"trial"&lt;/p&gt;&lt;script&gt;&lt;/script&gt;'
+
+    field.escape=True
+    
+    assert field.check('<p>"trial"</p><script></script>')=='&lt;p&gt;&quot;trial&quot;&lt;/p&gt;&lt;script&gt;&lt;/script&gt;'
+
+    field.trusted_tags=['p']
+    
+    assert field.check('<p>"trial"</p><script></script>')=='<p>&quot;trial&quot;</p>&lt;script&gt;&lt;/script&gt;'
+
+    #field.
+def test_test_foreignkeyfield():
+    
+    field=corefields.ForeignKeyField('foreign', ExampleModel())
+    
+    assert field.check('dsd')==None
+
+def test_test_booleanfield():
+    
+    field=corefields.BooleanField('bool')
+    
+    assert field.check('25')=='0'
+    
+    assert field.error==True
+    
+    
+def test_test_arrayfield():
+    
+    field=ArrayField('array', corefields.IntegerField('item'))
+    
+    assert field.check([1, 2, 3])=='["1", "2", "3"]'
+    
+    assert field.check([1, 2, '3t'])=='["1", "2", "0"]'
+    
+    assert field.error==True
+
+def test_test_colorfield():
+    
+    field=ColorField('color')
+    
+    assert field.check('#ff12f5')==0xff12f5
+    
+    assert field.check('#ff12f54')==0
+    
+    assert field.get_hex_color(0xaf42f5)=='#af42f5'
+    
+def test_test_datefield():
+    
+    datetime.timezone='Europe/Madrid'
+        
+    datetime.set_timezone()
+    
+    field=DateField('date')
+    
+    field.utc=False
+    
+    assert field.check('20201234121011')==''
+    
+    assert field.check('20201230121011')=='20201230121011'
+    
+def test_test_datetimefield():
+    
+    datetime.timezone='Europe/Madrid'
+        
+    datetime.set_timezone()
+    
+    field=DateTimeField('date')
+    
+    field.utc=False
+    
+    assert field.check('20201230121011')=='2020-12-30 12:10:11'
+    
+    assert field.check('20201290121011')=='0000-00-00 00:00:00'
+
+def test_test_dictfield():
+    
+    field=DictField('dict', corefields.IntegerField('item'))
+    
+    assert field.check({'come': '5'})=='{"come": "5"}'
+    
+    assert field.check({'come': '5t'})=='{"come": "0"}'
+
+    assert field.error==True
+
+def test_test_emailfield():
+    
+    field=EmailField('email')
+    
+    assert field.check('example@example.com')=='example@example.com'
+    
+    assert field.check('example-example.com')==''
+    
+def test_test_i18nfield():
+    
+    field=I18nField('i18nfield')
+    
+    assert field.check(json.dumps({'en-US': "Hello", 'es-ES': "Hola"}))=='{"en-US": "Hello", "es-ES": "Hola"}'
+    
+    assert field.check(json.dumps({'en-US': "Hello"}))=='{"en-US": "Hello", "es-ES": ""}'
+    
+    pass
+    

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': '', '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)
+