tests/test_wrfeed.py

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