Datenbank über unittest.

Readme.md

@@ -21,4 +21,5 @@ CC-BY-SA Olli Graf
 |14 | Logging|
 |15 | GPIO|
 |16 | Testing|
+|17 | Datenbank|
 

db/.gitignore

@@ -1,3 +1,2 @@
 ./__pycache__
-*.log
 ./network/__pycache__/*

db/DBConnection.py

@@ -0,0 +1,53 @@
+import mysql.connector
+import logging
+
+logging.basicConfig( format='%(asctime)-15s [%(levelname)s] %(funcName)s: %(message)s', level=logging.DEBUG)
+class DBConnection():
+
+  def __init__(self):
+    self.dbconnection = None
+
+  def connect(self, host, user, password, database):
+
+    self.connection= mysql.connector.connect(host=host, user=user,password=password,database=database)
+    logging.debug(f'Datenbankverbindung aufgebaut ${self.connection}')
+    logging.debug(f'Cursor gesetzt ${self.cursor}')
+
+  def disconnect(self):
+    if self.connection is not None:
+      self.connection.close()
+
+
+  def isConnected(self):
+#    logging.debug(f'connection={self.connection}')
+#    logging.debug(f'connected= {self.connection.is_connected()}')
+    if self.connection is not None:
+      return self.connection.is_connected()
+    return False
+
+  def cursor(self):
+     if self.isConnected():
+       return self.connection.cursor()
+     return None
+
+  def insert(self,cmd):
+    pass
+
+  def fetch(self):
+    pass  
+
+  def create_table(self):
+    pass
+
+  def execute_query(self, query, params = None):
+    if self.isConnected():
+      cursor = self.cursor()
+      if cursor is not None:
+        cursor.execute(query,params)
+        return cursor.fetchall()
+
+
+
+  
+
+

db/connectdb.py

@@ -1,10 +1,9 @@
-import mysql.connector
+from  DBConnection import DBConnection
 import logging
 
 logging.basicConfig( format='%(asctime)-15s [%(levelname)s] %(funcName)s: %(message)s', level=logging.DEBUG)
+db_connection = DBConnection()
 
-mydb = mysql.connector.connect(host='database', user='ploeger',password='SW_Ep.6:ROTJ',database='weather')
+db_connection.connect('database','piuser','SW_Ep.6:ROTJ','pidb')
-logging.debug(f'Datenbankverbindung aufgebaut ${mydb}')
+logging.debug(f'Datenbankverbindung aufgebaut ${db_connection.isConnected()}')
-mycursor = mydb.cursor()
-logging.debug(f'Cursor gesetzt ${mycursor}')
 

db/testdb.py

@@ -0,0 +1,65 @@
+import unittest
+from  DBConnection import DBConnection
+import logging
+
+logging.basicConfig( format='%(asctime)-15s [%(levelname)s] %(funcName)s: %(message)s', level=logging.DEBUG)
+
+class TestDBConnection(unittest.TestCase):
+
+  @classmethod
+  def setUpClass(cls):
+    logging.debug(f'setUpClass() {cls}')
+
+    cls.db_connection = DBConnection()
+    cls.db_connection.connect('database','piuser','SW_Ep.6:ROTJ','pidb')
+    logging.debug('erzeuge Table fib...')
+    cls.db_connection.execute_query('create table if not exists fib (id int auto_increment primary key, wert int)')
+    cls.db_connection.execute_query('create table if not exists schueler (id int auto_increment primary key, vorname varchar(20),name varchar(20))')
+    logging.debug('setup() fertig')
+
+  @classmethod
+  def tearDownClass(cls):
+    logging.debug(f'tearDownClass() {cls}')
+
+    logging.debug('Räume Tables auf')
+    cls.db_connection.execute_query('drop table if exists fib')
+    cls.db_connection.execute_query('drop table if exists schueler')
+    logging.debug('schliesse Datenbankverbindung.')
+    cls.db_connection.disconnect()
+    cls.db_connection = None
+
+
+  def test_insert(self):
+    logging.debug(f'connected= {self.db_connection.isConnected()}')
+    self.assertTrue(self.db_connection.isConnected())
+    self.db_connection.execute_query("insert into schueler (name,vorname) values('Simpson','Bart')")
+    self.db_connection.execute_query("insert into schueler (name,vorname) values('Simpson','Lisa')")
+    self.db_connection.execute_query("insert into schueler (name,vorname) values('van Houten','Milhouse')")
+    self.db_connection.execute_query("insert into schueler (name,vorname) values('Wiggum','Ralph')")
+#    input('Press Any Key')
+
+  def test_select(self):
+#    self.test_insert()
+    self.assertTrue(self.db_connection.isConnected())
+    result = self.db_connection.execute_query("select * from  schueler where name='Simpson'")
+    self.assertEqual(2,len(result))
+    result = self.db_connection.execute_query("select * from  schueler")
+    self.assertEqual(4,len(result))
+
+  def test_connected(self):
+    self.assertTrue(self.db_connection.isConnected())
+
+  def test_disconnect(self):
+    self.db_connection.disconnect()
+    self.assertFalse(self.db_connection.isConnected())
+
+  def test_create_table(self):
+    self.assertTrue(self.db_connection.isConnected())
+
+      
+    
+
+
+if __name__ =='__main__':
+  unittest.main()
+