diff --git a/db/testdb.py b/db/testdb.py deleted file mode 100644 index cb92846..0000000 --- a/db/testdb.py +++ /dev/null @@ -1,65 +0,0 @@ -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() - diff --git a/db/.gitignore b/teil17/.gitignore similarity index 80% rename from db/.gitignore rename to teil17/.gitignore index c7fb3aa..f1f4d92 100644 --- a/db/.gitignore +++ b/teil17/.gitignore @@ -1,2 +1,3 @@ ./__pycache__ +./db.ini ./network/__pycache__/* diff --git a/db/DBConnection.py b/teil17/DBConnection.py similarity index 54% rename from db/DBConnection.py rename to teil17/DBConnection.py index 3726c82..91d2e7d 100644 --- a/db/DBConnection.py +++ b/teil17/DBConnection.py @@ -1,3 +1,4 @@ +#encoding: utf-8 import mysql.connector import logging @@ -7,47 +8,38 @@ class DBConnection(): def __init__(self): self.dbconnection = None + #Verbindung zur Datenbank mit den übergebenen Wreten herstellen. 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}') + # Verbindung trennen def disconnect(self): if self.connection is not None: self.connection.close() + self.connection = None + # Schüler-Record in die Tabelle einfügen. + def insertSchueler(self,schueler): + self.execute_query('insert into schueler (name,vorname) values(%s,%s)', (schueler['name'],schueler['vorname'])) + # Liefert True, wenn die Verbindung zur DB besteht, sonst False 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 + # Wenn DB-Verbindung besteht, wird der Cursor geliefert, sonst None 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 - + # Datenbank Kommando ausführen. Das params-Tupel wird in die Query eingesetzt. + # Liefert ein Resultset zurück. 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() - - - - - - + cursor.execute(query,params) #Kommdo ausführen + return cursor.fetchall() # Ergebnis zurückliefern. diff --git a/teil17/__pycache__/DBConnection.cpython-39.pyc b/teil17/__pycache__/DBConnection.cpython-39.pyc new file mode 100644 index 0000000..6b2e1b2 Binary files /dev/null and b/teil17/__pycache__/DBConnection.cpython-39.pyc differ diff --git a/db/connectdb.py b/teil17/connectdb.py similarity index 100% rename from db/connectdb.py rename to teil17/connectdb.py diff --git a/teil17/db.ini.sample b/teil17/db.ini.sample new file mode 100644 index 0000000..41f7420 --- /dev/null +++ b/teil17/db.ini.sample @@ -0,0 +1,8 @@ +[pidb] +host=database +user=piuser +password= +database=pidb + + + diff --git a/teil17/testdb.py b/teil17/testdb.py new file mode 100755 index 0000000..4100704 --- /dev/null +++ b/teil17/testdb.py @@ -0,0 +1,88 @@ +#!/usr/bin/python3 +# encoding:utf-8 + +import unittest +from DBConnection import DBConnection +import logging +import configparser + +logging.basicConfig( format='%(asctime)-15s [%(levelname)s] %(funcName)s: %(message)s', level=logging.DEBUG) + +schueler = [ + {'name':'Simpson','vorname': 'Bart'}, + {'name':'Simpson','vorname':'Lisa'}, + {'name':'van Houten','vorname':'Milhouse'}, + {'name':'Wiggum','vorname':'Ralph'}, + {'name':'Jones','vorname':'Jimbo'} +] + +class TestDBConnection(unittest.TestCase): + + @classmethod + def setUpClass(cls): + logging.debug(f'setUpClass() {cls}') + + config = configparser.ConfigParser() + config.read('db.ini') + logging.debug(f"host={config['pidb']['host']}") + + cls.db_connection = DBConnection() + cls.db_connection.connect(config['pidb']['host'],config['pidb']['user'],config['pidb']['password'],config['pidb']['database']) + if cls.db_connection.isConnected(): + logging.debug('erzeuge Table schueler...') + cls.db_connection.execute_query('create table if not exists schueler (id int auto_increment primary key, vorname varchar(20),name varchar(20))') + + # Testdaten aufbauen. + cls.fill_table() + 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 schueler') + logging.debug('schliesse Datenbankverbindung.') + cls.db_connection.disconnect() + cls.db_connection = None + + +# input('Press Any Key') + + #Testdaten in die Datenbank übertragen + @classmethod + def fill_table(cls): + for s in range(0,len(schueler)): + student = schueler[s] + cls.db_connection.insertSchueler(student) + +# Nur die Records mit Nachname ='Simpson' lesen + def test_select_name(self): + self.assertTrue(self.db_connection.isConnected()) + + result = self.db_connection.execute_query("select * from schueler where name='Simpson'") + self.assertEqual(2,len(result)) + + def test_select_all(self): +# self.test_insert() + result = self.db_connection.execute_query("select * from schueler") + logging.debug(f'result={result}') + self.assertEqual(len(schueler),len(result)) + + for i in range(len(result)): + self.assertEqual(schueler[i]['vorname'], result[i][1]) + + def test_connected(self): + self.assertTrue(self.db_connection.isConnected()) + + + def test_create_table(self): + self.assertTrue(self.db_connection.isConnected()) + + + + + +if __name__ =='__main__': + unittest.main() +