Test logbook

.travis.yml

@@ -3,8 +3,13 @@ language: python
 python:
   - 3.4
 
+services:
+  - postgresql
+
 before_script:
   - flake8 tests ogn
+  - psql -c 'CREATE DATABASE ogn_test;' -U postgres
+  - psql -c 'CREATE EXTENSION postgis;' -U postgres -d travis_postgis
 
 script:
   - nosetests --with-coverage --cover-package=ogn

config/test.py

@@ -0,0 +1,9 @@
+SQLALCHEMY_DATABASE_URI = 'postgresql://postgres@localhost:5432/ogn_test'
+
+BROKER_URL = 'redis://localhost:6379/0'
+CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
+
+
+CELERYBEAT_SCHEDULE = {}
+
+CELERY_TIMEZONE = 'UTC'

tests/collect/test_logbook.py

@@ -0,0 +1,115 @@
+import unittest
+import os
+
+from ogn.model import Logbook
+
+from ogn.collect.logbook import compute_logbook_entries
+
+
+class TestDB(unittest.TestCase):
+    session = None
+    engine = None
+    app = None
+
+    def setUp(self):
+        os.environ['OGN_CONFIG_MODULE'] = 'config.test'
+        from ogn.commands.dbutils import engine, session
+        self.session = session
+        self.engine = engine
+
+        session.execute("INSERT INTO device(address) VALUES ('DD0815'), ('DD4711')")
+        session.execute("INSERT INTO airport(name) VALUES ('Koenigsdorf'), ('Ohlstadt')")
+
+    def tearDown(self):
+        session = self.session
+        session.execute("DELETE FROM takeoff_landing")
+        session.execute("DELETE FROM logbook")
+        session.execute("DELETE FROM device")
+        session.execute("DELETE FROM airport")
+        session.commit()
+        pass
+
+    def count_logbook_entries(self):
+        session = self.session
+        logbook_query = session.query(Logbook)
+        i = 0
+        for logbook in logbook_query.all():
+            i = i + 1
+            print("{} {} {} {} {} {}".format(logbook.id, logbook.device_id, logbook.takeoff_airport_id, logbook.takeoff_timestamp, logbook.landing_airport_id, logbook.landing_timestamp))
+
+        return i
+
+    def test_single_takeoff(self):
+        session = self.session
+
+        session.execute("INSERT INTO takeoff_landing(device_id, airport_id, timestamp, is_takeoff) SELECT d.id, a.id, '2016-06-01 10:00:00', TRUE FROM airport a, device d WHERE a.name='Koenigsdorf' and d.address = 'DD0815'")
+        session.commit()
+
+        compute_logbook_entries(session)
+        self.assertEqual(self.count_logbook_entries(), 1)
+
+    def test_single_landing(self):
+        session = self.session
+
+        session.execute("INSERT INTO takeoff_landing(device_id, airport_id, timestamp, is_takeoff) SELECT d.id, a.id, '2016-06-01 10:00:00', FALSE FROM airport a, device d WHERE a.name='Koenigsdorf' and d.address = 'DD0815'")
+        session.commit()
+
+        compute_logbook_entries(session)
+        self.assertEqual(self.count_logbook_entries(), 1)
+
+    def test_different_takeoffs(self):
+        session = self.session
+
+        session.execute("INSERT INTO takeoff_landing(device_id, airport_id, timestamp, is_takeoff) SELECT d.id, a.id, '2016-06-01 10:00:00', TRUE FROM airport a, device d WHERE a.name='Koenigsdorf' and d.address = 'DD0815'")
+        session.execute("INSERT INTO takeoff_landing(device_id, airport_id, timestamp, is_takeoff) SELECT d.id, a.id, '2016-06-01 10:00:00', TRUE FROM airport a, device d WHERE a.name='Ohlstadt' and d.address = 'DD4711'")
+        session.commit()
+
+        compute_logbook_entries(session)
+        self.assertEqual(self.count_logbook_entries(), 2)
+
+    def test_takeoff_and_landing(self):
+        session = self.session
+
+        session.execute("INSERT INTO takeoff_landing(device_id, airport_id, timestamp, is_takeoff) SELECT d.id, a.id, '2016-06-01 10:00:00', TRUE FROM airport a, device d WHERE a.name='Koenigsdorf' and d.address = 'DD0815'")
+        session.execute("INSERT INTO takeoff_landing(device_id, airport_id, timestamp, is_takeoff) SELECT d.id, a.id, '2016-06-01 10:05:00', FALSE FROM airport a, device d WHERE a.name='Koenigsdorf' and d.address = 'DD0815'")
+        session.commit()
+
+        compute_logbook_entries(session)
+        self.assertEqual(self.count_logbook_entries(), 1)
+
+    def test_takeoff_and_landing_on_different_days(self):
+        session = self.session
+
+        session.execute("INSERT INTO takeoff_landing(device_id, airport_id, timestamp, is_takeoff) SELECT d.id, a.id, '2016-06-01 10:00:00', TRUE FROM airport a, device d WHERE a.name='Koenigsdorf' and d.address = 'DD0815'")
+        session.execute("INSERT INTO takeoff_landing(device_id, airport_id, timestamp, is_takeoff) SELECT d.id, a.id, '2016-06-02 10:05:00', FALSE FROM airport a, device d WHERE a.name='Koenigsdorf' and d.address = 'DD0815'")
+        session.commit()
+
+        compute_logbook_entries(session)
+        self.assertEqual(self.count_logbook_entries(), 2)
+
+    def test_update(self):
+        session = self.session
+
+        session.execute("INSERT INTO takeoff_landing(device_id, airport_id, timestamp, is_takeoff) SELECT d.id, a.id, '2016-06-01 10:00:00', TRUE FROM airport a, device d WHERE a.name='Koenigsdorf' and d.address = 'DD0815'")
+        session.commit()
+        compute_logbook_entries(session)
+        session.execute("INSERT INTO takeoff_landing(device_id, airport_id, timestamp, is_takeoff) SELECT d.id, a.id, '2016-06-01 10:05:00', FALSE FROM airport a, device d WHERE a.name='Koenigsdorf' and d.address = 'DD0815'")
+        session.commit()
+        compute_logbook_entries(session)
+
+        self.assertEqual(self.count_logbook_entries(), 1)
+
+    def test_update_wrong_order(self):
+        session = self.session
+
+        session.execute("INSERT INTO takeoff_landing(device_id, airport_id, timestamp, is_takeoff) SELECT d.id, a.id, '2016-06-01 10:05:00', FALSE FROM airport a, device d WHERE a.name='Koenigsdorf' and d.address = 'DD0815'")
+        session.commit()
+        compute_logbook_entries(session)
+        session.execute("INSERT INTO takeoff_landing(device_id, airport_id, timestamp, is_takeoff) SELECT d.id, a.id, '2016-06-01 10:00:00', TRUE FROM airport a, device d WHERE a.name='Koenigsdorf' and d.address = 'DD0815'")
+        session.commit()
+        compute_logbook_entries(session)
+
+        self.assertEqual(self.count_logbook_entries(), 1)
+
+if __name__ == '__main__':
+    unittest.main()