tests/test_wrmwdb.py

   1 #!/usr/bin/python3.4
   2 import os
   3 import unittest
   4 import sqlalchemy
   5 from sqlalchemy.engine import create_engine
   6 from sqlalchemy.orm import sessionmaker
   7 import wrpylib.wrmwdb
   8
   9
  10 TEST_DB = 'mysql://philipp@localhost:3306/philipp_winterrodeln_wrpylib_wrpylib_test?charset=utf8&use_unicode=1'
  11
  12
  13 def setup_test_database():
  14     engine = create_engine(TEST_DB)
  15     testdb = os.path.join(os.path.dirname(__file__), 'testdb.sql')
  16     with open(testdb, 'r') as f:
  17         sql = f.read()
  18     with engine.begin() as con:
  19         con.execute(sql)
  20
  21         # update dates
  22         seconds_diff, = con.execute("SELECT TIMESTAMPDIFF(SECOND, '2017-03-31 07:00', now())").fetchone()
  23         seconds_diff = int(seconds_diff)
  24         con.execute(
  25             "UPDATE wrreport SET date_report = DATE(date_report + INTERVAL %s SECOND) WHERE time_report IS NULL",
  26             (seconds_diff,))
  27         con.execute(
  28             "UPDATE wrreport SET date_report = DATE(TIMESTAMP(date_report, time_report) + INTERVAL %s SECOND) WHERE time_report IS NOT NULL",
  29             (seconds_diff,))
  30         con.execute(
  31             "UPDATE wrreport SET time_report = TIME(TIMESTAMP(date_report, time_report) + INTERVAL %s SECOND) WHERE time_report IS NOT NULL",
  32             (seconds_diff,))
  33         con.execute("UPDATE wrreport SET date_entry = date_entry + INTERVAL %s SECOND", (seconds_diff,))
  34         con.execute("UPDATE wrreport SET date_invalid = date_invalid + INTERVAL %s SECOND", (seconds_diff,))
  35         con.execute("UPDATE wrreport SET delete_date = delete_date + INTERVAL %s SECOND", (seconds_diff,))
  36     return engine
  37
  38
  39 class TestWrMwDb(unittest.TestCase):
  40     @classmethod
  41     def setUpClass(cls):
  42         cls.engine = setup_test_database()
  43         cls.Session = sessionmaker(bind=cls.engine)
  44
  45     def setUp(self):
  46         self.metadata = sqlalchemy.MetaData()
  47         self.session = self.Session()
  48
  49     def tearDown(self):
  50         self.session.rollback()
  51
  52     def test_wrreport_table(self):
  53         WrReport = wrpylib.wrmwdb.wrreport_table(self.metadata)
  54         wrreport = self.session.query(WrReport).filter(WrReport.c.id==19590).first()
  55         self.assertEqual(wrreport.id, 19590)
  56         self.assertEqual('Pleisenhütte Axamer Lizum', wrreport.page_title)
  57
  58     def test_wrsledruncache_table(self):
  59         WrSledrunCache = wrpylib.wrmwdb.wrsledruncache_table(self.metadata)
  60         wrsledruncache = self.session.query(WrSledrunCache).filter(WrSledrunCache.c.page_id==926).first()
  61         self.assertEqual(926, wrsledruncache.page_id)
  62         self.assertEqual('Hühnerspiel', wrsledruncache.page_title)
  63
  64     def test_wrinncache_table(self):
  65         WrInnCache = wrpylib.wrmwdb.wrinncache_table(self.metadata)
  66         wrinncache = self.session.query(WrInnCache).filter(WrInnCache.c.page_id==362).first()
  67         self.assertEqual(362, wrinncache.page_id)
  68         self.assertEqual('Meißner_Haus_(Gasthaus)', wrinncache.page_title)
  69
  70     def test_wrreportcache_table(self):
  71         WrReportCache = wrpylib.wrmwdb.wrreportcache_table(self.metadata)
  72         wrreport = self.session.query(WrReportCache).filter(WrReportCache.c.page_id==226).first()
  73         self.assertEqual(226, wrreport.page_id)
  74         self.assertEqual('Axamer Lizum', wrreport.page_title)
  75
  76     def test_wrregion_table(self):
  77         WrRegion = wrpylib.wrmwdb.wrregion_table(self.metadata)
  78         wrregion = self.session.query(WrRegion).filter(WrRegion.c.id==18).first()
  79         self.assertEqual(18, wrregion.id)
  80         self.assertEqual('Salzburg', wrregion.name)
  81
  82     def test_wrregioncache_table(self):
  83         WrRegionCache = wrpylib.wrmwdb.wrregioncache_table(self.metadata)
  84         wrregioncache = self.session.query(WrRegionCache).filter((WrRegionCache.c.region_id==27) & (WrRegionCache.c.page_id==5)).first()
  85         self.assertEqual(27, wrregioncache.region_id)
  86         self.assertEqual(5, wrregioncache.page_id)