import os import unittest import sqlalchemy from sqlalchemy.engine import create_engine from sqlalchemy.orm import sessionmaker import wrpylib.wrmwdb from wrpylib.wrmwcache import update_wrreportcache TEST_DB = 'mysql://philipp@localhost:3306/philipp_winterrodeln_wrpylib_wrpylib_test?charset=utf8mb4' def setup_test_database(): engine = create_engine(TEST_DB) testdb = os.path.join(os.path.dirname(__file__), 'testdb.sql') with open(testdb, 'r') as f: sql = f.read() with engine.begin() as con: con.execution_options(no_parameters=True).execute(sql) # update dates seconds_diff, = con.execute("SELECT TIMESTAMPDIFF(SECOND, '2017-03-31 07:00', now())").fetchone() seconds_diff = int(seconds_diff) con.execute( "UPDATE wrreport SET date_report = DATE(date_report + INTERVAL %s SECOND) WHERE time_report IS NULL", (seconds_diff,)) con.execute( "UPDATE wrreport SET date_report = DATE(TIMESTAMP(date_report, time_report) + INTERVAL %s SECOND) WHERE time_report IS NOT NULL", (seconds_diff,)) con.execute( "UPDATE wrreport SET time_report = TIME(TIMESTAMP(date_report, time_report) + INTERVAL %s SECOND) WHERE time_report IS NOT NULL", (seconds_diff,)) con.execute("UPDATE wrreport SET date_entry = date_entry + INTERVAL %s SECOND", (seconds_diff,)) con.execute("UPDATE wrreport SET date_invalid = date_invalid + INTERVAL %s SECOND", (seconds_diff,)) con.execute("UPDATE wrreport SET delete_date = delete_date + INTERVAL %s SECOND", (seconds_diff,)) return engine class TestWrMwDb(unittest.TestCase): @classmethod def setUpClass(cls): cls.engine = setup_test_database() cls.Session = sessionmaker(bind=cls.engine) def setUp(self): self.metadata = sqlalchemy.MetaData() self.session = self.Session() def tearDown(self): self.session.rollback() def test_wrreport_table(self): WrReport = wrpylib.wrmwdb.wrreport_table(self.metadata) wrreport = self.session.query(WrReport).filter(WrReport.c.id==19590).first() self.assertEqual(wrreport.id, 19590) self.assertEqual('Pleisenhütte Axamer Lizum', wrreport.page_title) def test_wrsledruncache_table(self): WrSledrunCache = wrpylib.wrmwdb.wrsledruncache_table(self.metadata) wrsledruncache = self.session.query(WrSledrunCache).filter(WrSledrunCache.c.page_id==926).first() self.assertEqual(926, wrsledruncache.page_id) self.assertEqual('Hühnerspiel', wrsledruncache.page_title) def test_wrinncache_table(self): WrInnCache = wrpylib.wrmwdb.wrinncache_table(self.metadata) wrinncache = self.session.query(WrInnCache).filter(WrInnCache.c.page_id==362).first() self.assertEqual(362, wrinncache.page_id) self.assertEqual('Meißner_Haus_(Gasthaus)', wrinncache.page_title) def test_wrreportcache_table(self): update_wrreportcache(self.engine.connect()) WrReportCache = wrpylib.wrmwdb.wrreportcache_table(self.metadata) wrreport = self.session.query(WrReportCache).filter(WrReportCache.c.page_id == 237).first() self.assertEqual(237, wrreport.page_id) self.assertEqual('Gleirschalm', wrreport.page_title) def test_wrregion_table(self): WrRegion = wrpylib.wrmwdb.wrregion_table(self.metadata) wrregion = self.session.query(WrRegion).filter(WrRegion.c.id == 18).first() self.assertEqual(18, wrregion.id) self.assertEqual('Salzburg', wrregion.name) def test_wrregioncache_table(self): WrRegionCache = wrpylib.wrmwdb.wrregioncache_table(self.metadata) wrregioncache = self.session.query(WrRegionCache).filter((WrRegionCache.c.region_id == 27) & (WrRegionCache.c.page_id == 5)).first() self.assertEqual(27, wrregioncache.region_id) self.assertEqual(5, wrregioncache.page_id)