mirror
/
ogn-python
kopia lustrzana https://github.com/glidernet/ogn-python
1
0
Forkuj 0
Kod Zgłoszenia Projekty Wydania Wiki Aktywność

Better logbook testing and refactoring

pull/68/head
Konstantin Gründger 2017-12-16 15:54:04 +01:00
rodzic a5da270d6d
commit 307278221a
4 zmienionych plików z 113 dodań i 51 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

3
ogn/model/airport.py
Wyświetl plik

@ -25,6 +25,9 @@ class Airport(Base):
# Relations
takeoff_landings = relationship('TakeoffLanding')
def __init__(self, name):
self.name = name
def __repr__(self):
return "<Airport %s: %s,%s,%s,%s,%s,%s,%s,%s,%s,% s>" % (
self.name,

3
ogn/model/device.py
Wyświetl plik

@ -22,6 +22,9 @@ class Device(Base):
# Relations
aircraft_beacons = relationship('AircraftBeacon')
def __init__(self, address):
self.address = address
def __repr__(self):
return "<Device: %s,%s,%s,%s,%s,%s>" % (
self.address,

8
ogn/model/takeoff_landing.py
Wyświetl plik

@ -9,9 +9,9 @@ class TakeoffLanding(Base):
id = Column(Integer, primary_key=True)
is_takeoff = Column(Boolean)
timestamp = Column(DateTime, index=True)
track = Column(Integer)
is_takeoff = Column(Boolean)
# Relations
airport_id = Column(Integer, ForeignKey('airport.id', ondelete='SET NULL'), index=True)
@ -19,3 +19,9 @@ class TakeoffLanding(Base):
device_id = Column(Integer, ForeignKey('device.id', ondelete='SET NULL'), index=True)
device = relationship('Device', foreign_keys=[device_id])
def __init__(self, is_takeoff, timestamp, airport_id, device_id):
self.is_takeoff = is_takeoff
self.timestamp = timestamp
self.airport_id = airport_id
self.device_id = device_id

150
tests/collect/test_logbook.py
Wyświetl plik

@ -1,10 +1,11 @@
import unittest
import os
from sqlalchemy.sql import null, and_
from sqlalchemy.sql import null
from ogn.model import Logbook, Airport
from ogn.model import Logbook, Airport, Device
from ogn.collect.logbook import update_logbook
from ogn.model.takeoff_landing import TakeoffLanding
class TestDB(unittest.TestCase):
@ -12,11 +13,6 @@ class TestDB(unittest.TestCase):
engine = None
app = None
TAKEOFF_KOENIGSDF_DD0815 = "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'"
LANDING_KOENIGSDF_DD0815 = "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'"
LANDING_KOENIGSDF_DD0815_LATER = "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'"
TAKEOFF_OHLSTADT_DD4711 = "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'"
def setUp(self):
os.environ['OGN_CONFIG_MODULE'] = 'config.test'
from ogn.commands.dbutils import engine, session
@ -26,8 +22,25 @@ class TestDB(unittest.TestCase):
from ogn.commands.database import init
init()
session.execute("INSERT INTO device(address) VALUES ('DD0815'), ('DD4711')")
session.execute("INSERT INTO airport(name) VALUES ('Koenigsdorf'), ('Ohlstadt')")
# Create basic data and insert
self.dd0815 = Device(address='DD0815')
self.dd4711 = Device(address='DD4711')
self.koenigsdorf = Airport(name='Koenigsdorf')
self.ohlstadt = Airport(name='Ohlstadt')
session.add(self.dd0815)
session.add(self.dd4711)
session.add(self.koenigsdorf)
session.add(self.ohlstadt)
session.commit()
# Prepare takeoff and landings
self.takeoff_koenigsdorf_dd0815 = TakeoffLanding(is_takeoff=True, timestamp='2016-06-01 10:00:00', airport_id=self.koenigsdorf.id, device_id=self.dd0815.id)
self.landing_koenigsdorf_dd0815 = TakeoffLanding(is_takeoff=False, timestamp='2016-06-01 10:05:00', airport_id=self.koenigsdorf.id, device_id=self.dd0815.id)
self.landing_koenigsdorf_dd0815_later = TakeoffLanding(is_takeoff=False, timestamp='2016-06-02 10:05:00', airport_id=self.koenigsdorf.id, device_id=self.dd0815.id)
self.takeoff_ohlstadt_dd4711 = TakeoffLanding(is_takeoff=True, timestamp='2016-06-01 10:00:00', airport_id=self.ohlstadt.id, device_id=self.dd4711.id)
def tearDown(self):
session = self.session
@ -36,135 +49,172 @@ class TestDB(unittest.TestCase):
session.execute("DELETE FROM device")
session.execute("DELETE FROM airport")
session.commit()
pass
def count_logbook_entries(self):
def get_logbook_entries(self):
session = self.session
query = session.query(Logbook)
return len(query.all())
return session.query(Logbook).order_by(Logbook.takeoff_airport_id, Logbook.reftime).all()
def assert_entries(self, koen_to=0, koen_ldg=0, koen_complete=0, ohl_to=0, ohl_ldg=0, ohl_complete=0):
session = self.session
entries = len(session.query(Logbook).filter(and_(Airport.id == Logbook.takeoff_airport_id, Airport.name == 'Koenigsdorf')).filter(Logbook.landing_airport_id == null()).all())
entries = len(session.query(Logbook).filter(Logbook.takeoff_airport_id == self.koenigsdorf.id).filter(Logbook.landing_airport_id == null()).all())
self.assertEqual(entries, koen_to)
entries = len(session.query(Logbook).filter(and_(Airport.id == Logbook.landing_airport_id, Airport.name == 'Koenigsdorf')).filter(Logbook.takeoff_airport_id == null()).all())
entries = len(session.query(Logbook).filter(Logbook.landing_airport_id == self.koenigsdorf.id).filter(Logbook.takeoff_airport_id == null()).all())
self.assertEqual(entries, koen_ldg)
entries = len(session.query(Logbook).filter(and_(Airport.id == Logbook.takeoff_airport_id, Airport.name == 'Koenigsdorf')).filter(Logbook.takeoff_airport_id == Logbook.landing_airport_id).all())
entries = len(session.query(Logbook).filter(Logbook.takeoff_airport_id == Logbook.landing_airport_id == self.koenigsdorf.id).all())
self.assertEqual(entries, koen_complete)
entries = len(session.query(Logbook).filter(and_(Airport.id == Logbook.takeoff_airport_id, Airport.name == 'Ohlstadt')).filter(Logbook.landing_airport_id == null()).all())
entries = len(session.query(Logbook).filter(Logbook.takeoff_airport_id == self.ohlstadt.id).filter(Logbook.landing_airport_id == null()).all())
self.assertEqual(entries, ohl_to)
entries = len(session.query(Logbook).filter(and_(Airport.id == Logbook.landing_airport_id, Airport.name == 'Ohlstadt')).filter(Logbook.takeoff_airport_id == null()).all())
entries = len(session.query(Logbook).filter(Logbook.landing_airport_id == self.ohlstadt.id).filter(Logbook.takeoff_airport_id == null()).all())
self.assertEqual(entries, ohl_ldg)
entries = len(session.query(Logbook).filter(and_(Airport.id == Logbook.takeoff_airport_id, Airport.name == 'Ohlstadt')).filter(Logbook.takeoff_airport_id == Logbook.landing_airport_id).all())
entries = len(session.query(Logbook).filter(Logbook.takeoff_airport_id == Logbook.landing_airport_id == self.ohlstadt.id).all())
self.assertEqual(entries, ohl_complete)
def test_single_takeoff(self):
session = self.session
session.execute(self.TAKEOFF_KOENIGSDF_DD0815)
session.add(self.takeoff_koenigsdorf_dd0815)
session.commit()
update_logbook(session)
self.assert_entries(koen_to=1)
entries = self.get_logbook_entries()
self.assertEqual(len(entries), 1)
self.assertEqual(entries[0].takeoff_airport_id, self.koenigsdorf.id)
self.assertEqual(entries[0].landing_airport_id, None)
update_logbook(session)
self.assert_entries(koen_to=1)
entries2 = self.get_logbook_entries()
self.assertEqual(entries, entries2)
def test_single_landing(self):
session = self.session
session.execute(self.LANDING_KOENIGSDF_DD0815)
session.add(self.landing_koenigsdorf_dd0815)
session.commit()
update_logbook(session)
self.assert_entries(koen_ldg=1)
entries = self.get_logbook_entries()
self.assertEqual(len(entries), 1)
self.assertEqual(entries[0].takeoff_airport_id, None)
self.assertEqual(entries[0].landing_airport_id, self.koenigsdorf.id)
update_logbook(session)
self.assert_entries(koen_ldg=1)
entries2 = self.get_logbook_entries()
self.assertEqual(entries, entries2)
def test_different_takeoffs(self):
session = self.session
session.execute(self.TAKEOFF_KOENIGSDF_DD0815)
session.execute(self.TAKEOFF_OHLSTADT_DD4711)
session.add(self.takeoff_koenigsdorf_dd0815)
session.add(self.takeoff_ohlstadt_dd4711)
session.commit()
update_logbook(session)
self.assert_entries(koen_to=1, ohl_to=1)
entries = self.get_logbook_entries()
self.assertEqual(len(entries), 2)
self.assertEqual(entries[0].takeoff_airport_id, self.koenigsdorf.id)
self.assertEqual(entries[1].takeoff_airport_id, self.ohlstadt.id)
update_logbook(session)
self.assert_entries(koen_to=1, ohl_to=1)
entries2 = self.get_logbook_entries()
self.assertEqual(entries, entries2)
def test_takeoff_and_landing(self):
session = self.session
session.execute(self.TAKEOFF_KOENIGSDF_DD0815)
session.execute(self.LANDING_KOENIGSDF_DD0815)
session.add(self.takeoff_koenigsdorf_dd0815)
session.add(self.landing_koenigsdorf_dd0815)
session.commit()
update_logbook(session)
self.assert_entries(koen_complete=1)
entries = self.get_logbook_entries()
self.assertEqual(len(entries), 1)
self.assertEqual(entries[0].takeoff_airport_id, self.koenigsdorf.id)
self.assertEqual(entries[0].landing_airport_id, self.koenigsdorf.id)
update_logbook(session)
self.assert_entries(koen_complete=1)
entries2 = self.get_logbook_entries()
self.assertEqual(entries, entries2)
def test_takeoff_and_landing_on_different_days(self):
session = self.session
session.execute(self.TAKEOFF_KOENIGSDF_DD0815)
session.execute(self.LANDING_KOENIGSDF_DD0815_LATER)
session.add(self.takeoff_koenigsdorf_dd0815)
session.add(self.landing_koenigsdorf_dd0815_later)
session.commit()
update_logbook(session)
self.assert_entries(koen_to=1, koen_ldg=1)
entries = self.get_logbook_entries()
self.assertEqual(len(entries), 2)
self.assertEqual(entries[0].takeoff_airport_id, self.koenigsdorf.id)
self.assertEqual(entries[0].reftime, self.takeoff_koenigsdorf_dd0815.timestamp)
self.assertEqual(entries[1].landing_airport_id, self.koenigsdorf.id)
self.assertEqual(entries[1].reftime, self.landing_koenigsdorf_dd0815_later.timestamp)
update_logbook(session)
self.assert_entries(koen_to=1, koen_ldg=1)
entries2 = self.get_logbook_entries()
self.assertEqual(entries, entries2)
def test_update(self):
session = self.session
session.execute(self.TAKEOFF_KOENIGSDF_DD0815)
session.add(self.takeoff_koenigsdorf_dd0815)
session.commit()
update_logbook(session)
self.assert_entries(koen_to=1)
entries = self.get_logbook_entries()
self.assertEqual(len(entries), 1)
self.assertEqual(entries[0].takeoff_airport_id, self.koenigsdorf.id)
session.execute(self.LANDING_KOENIGSDF_DD0815)
session.add(self.landing_koenigsdorf_dd0815)
session.commit()
update_logbook(session)
self.assert_entries(koen_complete=1)
entries = self.get_logbook_entries()
self.assertEqual(len(entries), 1)
self.assertEqual(entries[0].takeoff_airport_id, self.koenigsdorf.id)
self.assertEqual(entries[0].landing_airport_id, self.koenigsdorf.id)
session.execute(self.TAKEOFF_OHLSTADT_DD4711)
session.add(self.takeoff_ohlstadt_dd4711)
session.commit()
update_logbook(session)
self.assert_entries(koen_complete=1, ohl_to=1)
update_logbook(session)
self.assert_entries(koen_complete=1, ohl_to=1)
entries = self.get_logbook_entries()
self.assertEqual(len(entries), 2)
self.assertEqual(entries[0].takeoff_airport_id, self.koenigsdorf.id)
self.assertEqual(entries[1].takeoff_airport_id, self.ohlstadt.id)
def test_update_wrong_order(self):
session = self.session
session.execute(self.LANDING_KOENIGSDF_DD0815)
session.add(self.landing_koenigsdorf_dd0815)
session.commit()
update_logbook(session)
self.assert_entries(koen_ldg=1)
entries = self.get_logbook_entries()
self.assertEqual(len(entries), 1)
self.assertEqual(entries[0].takeoff_airport_id, None)
self.assertEqual(entries[0].landing_airport_id, self.koenigsdorf.id)
self.assertEqual(entries[0].reftime, self.landing_koenigsdorf_dd0815.timestamp)
session.execute(self.TAKEOFF_KOENIGSDF_DD0815)
session.add(self.takeoff_koenigsdorf_dd0815)
session.commit()
update_logbook(session)
self.assert_entries(koen_complete=1)
entries = self.get_logbook_entries()
self.assertEqual(len(entries), 1)
self.assertEqual(entries[0].takeoff_airport_id, self.koenigsdorf.id)
self.assertEqual(entries[0].landing_airport_id, self.koenigsdorf.id)
self.assertEqual(entries[0].reftime, self.takeoff_koenigsdorf_dd0815.timestamp)
update_logbook(session)
entries2 = self.get_logbook_entries()
self.assertEqual(entries, entries2)
if __name__ == '__main__':