tests/test_wrfeed.py

   1 """How to test wrfeed:
   2
   3 WRFEED_SETTINGS=../wrfeed/test.cfg python3 -m unittest tests/test_wrfeed.py
   4 """
   5 import unittest
   6 import xml.etree.ElementTree as et
   7 import sqlalchemy
   8 from sqlalchemy.engine.url import URL, make_url
   9 import wrfeed
  10
  11
  12 class WrfeedTestBase(unittest.TestCase):
  13     def setUp(self):
  14         wrfeed.app.config['TESTING'] = True
  15         self.app = wrfeed.app.test_client()
  16
  17
  18 class TestNoDbWrfeed(WrfeedTestBase):
  19     def test_root(self):
  20         result = self.app.get('/')
  21         self.assertTrue(result.data.startswith(b'<?xml version='))
  22         self.assertIn(b'page about feeds', result.data)
  23
  24
  25 class TestDbWrfeed(WrfeedTestBase):
  26     def setUp(self):
  27         super().setUp()
  28         with wrfeed.app.app_context():
  29             # create database
  30             uri = make_url(wrfeed.app.config['DATABASE_URI'])
  31             self.uri = uri
  32             uri_nodb = URL(uri.drivername, host=uri.host, port=uri.port, username=uri.username, password=uri.password, query=uri.query)
  33             engine_nodb = sqlalchemy.create_engine(uri_nodb)
  34             engine_nodb.execute("DROP DATABASE IF EXISTS {}".format(uri.database))
  35             engine_nodb.execute("CREATE DATABASE {} CHARACTER SET = '{}'".format(uri.database, 'UTF8'))
  36
  37             # connect to database
  38             self.engine = sqlalchemy.create_engine(uri)
  39             con = self.engine.connect()
  40
  41             # fill database
  42             with open('tests/testdb.sql', 'r') as f:
  43                 sql = f.read()
  44             con.execute(sql)
  45
  46             # update dates
  47             seconds_diff, = con.execute("SELECT TIMESTAMPDIFF(SECOND, '2017-03-31 07:00', now())").fetchone()
  48             seconds_diff = int(seconds_diff)
  49             con.execute("UPDATE wrreport SET date_report = DATE(date_report + INTERVAL %s SECOND) WHERE time_report IS NULL", (seconds_diff,))
  50             con.execute("UPDATE wrreport SET date_report = DATE(TIMESTAMP(date_report, time_report) + INTERVAL %s SECOND) WHERE time_report IS NOT NULL", (seconds_diff,))
  51             con.execute("UPDATE wrreport SET time_report = TIME(TIMESTAMP(date_report, time_report) + INTERVAL %s SECOND) WHERE time_report IS NOT NULL", (seconds_diff,))
  52             con.execute("UPDATE wrreport SET date_entry = date_entry + INTERVAL %s SECOND", (seconds_diff,))
  53             con.execute("UPDATE wrreport SET date_invalid = date_invalid + INTERVAL %s SECOND", (seconds_diff,))
  54             con.execute("UPDATE wrreport SET delete_date = delete_date + INTERVAL %s SECOND", (seconds_diff,))
  55
  56
  57     def tearDown(self):
  58         with wrfeed.app.app_context():
  59             self.engine.execute("DROP DATABASE {}".format(self.uri.database))
  60
  61     def test_alle(self):
  62         result = self.app.get('/berichte/alle')
  63         self.assertTrue(result.data.startswith(b'<?xml'))
  64         root = et.fromstring(result.data)
  65         self.assertIn('Rodelbahnberichte', root.find('{http://www.w3.org/2005/Atom}title').text)
  66         self.assertEqual(len(root.findall('{http://www.w3.org/2005/Atom}entry')), 7)
  67
  68     def test_bahn(self):
  69         bahn_count = {
  70             'haunold': 1,
  71             'pleisenhütte_axamer_lizum': 2,
  72             'prantner_alm': 0,
  73             'birgitzer_alm_(vom_adelshof)': 1,
  74         }
  75         for bahn, count in bahn_count.items():
  76             with self.subTest(bahn=bahn):
  77                 result = self.app.get('/berichte/bahn/{}'.format(bahn))
  78                 self.assertTrue(result.data.startswith(b'<?xml'))
  79                 root = et.fromstring(result.data)
  80                 self.assertIn('Rodelbahnberichte', root.find('{http://www.w3.org/2005/Atom}title').text)
  81                 self.assertEqual(len(root.findall('{http://www.w3.org/2005/Atom}entry')), count, '"{}" not found {} times'.format(bahn, count))
  82
  83     def test_bahnen(self):
  84         bahnen_count = {
  85             '761': 2,
  86             '793': 0,
  87             '100': 0,
  88             '1417': 1,
  89             '5': 1,
  90             '228': 1,
  91             '793+100': 0,
  92             '761+1417+5+228': 5,
  93         }
  94         for bahnen, count in bahnen_count.items():
  95             with self.subTest(bahnen=bahnen):
  96                 result = self.app.get('/berichte/bahnen/{}'.format(bahnen))
  97                 self.assertTrue(result.data.startswith(b'<?xml'))
  98                 root = et.fromstring(result.data)
  99                 self.assertIn('Rodelbahnberichte', root.find('{http://www.w3.org/2005/Atom}title').text)
 100                 self.assertEqual(len(root.findall('{http://www.w3.org/2005/Atom}entry')), count, 'bahnen "{}" not found {} times'.format(bahnen, count))
 101
 102     def test_region(self):
 103         region_count = {
 104             'tirol': 6,
 105             'schweiz': 1,
 106             'innsbruck': 3,
 107         }
 108         for region, count in region_count.items():
 109             with self.subTest(region=region):
 110                 result = self.app.get('/berichte/region/{}'.format(region))
 111                 self.assertTrue(result.data.startswith(b'<?xml'))
 112                 root = et.fromstring(result.data)
 113                 self.assertIn('Rodelbahnberichte', root.find('{http://www.w3.org/2005/Atom}title').text)
 114                 self.assertEqual(len(root.findall('{http://www.w3.org/2005/Atom}entry')), count, '"{}" not found {} times'.format(region, count))