rattail/trainwreck/index

``rattail.trainwreck.handler``

==============================

.. automodule:: rattail.trainwreck.handler

   :members:

``rattail.trainwreck``

======================

.. automodule:: rattail.trainwreck

   :members:

.. toctree::

   :maxdepth: 1

   :caption: Contents:

   handler

#  Copyright © 2010-2022 Lance Edgar

    def get_trainwreck_handler(self, **kwargs):

        """

        Get the configured "trainwreck" handler.

        :returns: The :class:`~rattail.trainwreck.handler.TrainwreckHandler`

           instance for the app.

        """

        if not hasattr(self, 'trainwreck_handler'):

            spec = self.config.get('trainwreck', 'handler',

                                   default='rattail.trainwreck.handler:TrainwreckHandler')

            Handler = self.load_object(spec)

            self.trainwreck_handler = Handler(self.config)

        return self.trainwreck_handler

# -*- coding: utf-8; -*-

################################################################################

#

#  Rattail -- Retail Software Framework

#  Copyright © 2010-2022 Lance Edgar

#

#  This file is part of Rattail.

#

#  Rattail is free software: you can redistribute it and/or modify it under the

#  terms of the GNU General Public License as published by the Free Software

#  Foundation, either version 3 of the License, or (at your option) any later

#  version.

#

#  Rattail is distributed in the hope that it will be useful, but WITHOUT ANY

#  WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS

#  FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more

#  details.

#

#  You should have received a copy of the GNU General Public License along with

#  Rattail.  If not, see <http://www.gnu.org/licenses/>.

#

################################################################################

"""

Trainwreck Handler

"""

from __future__ import unicode_literals, absolute_import

import warnings

from rattail.app import GenericHandler

from rattail.util import OrderedDict

class TrainwreckHandler(GenericHandler):

    """

    Handler for Trainwreck data and databases.

    """

    def get_trainwreck_engines(self, include_hidden=True):

        """

        Return an "ordered" dict with configured trainwreck DB

        engines.  Keys of the dict will correspond to the config keys

        for each DB, values will be the engines.

        :param include_hidden: Flag indicating whether the result

           should include engines which are marked as hidden.  Note

           that hidden engines *are* included by default.

        :returns: An :class:`~rattail.util.OrderedDict` instance.

        """

        engines = OrderedDict(self.config.trainwreck_engines)

        if not include_hidden:

            for key in self.get_hidden_engine_keys():

                engines.pop(key, None)

        return engines

    def get_hidden_engine_keys(self):

        """

        Return a list of database engine keys which are configured to

        be hidden from the user interface.

        """

        hidden = self.config.getlist('trainwreck.db', 'hide',

                                     default=None)

        if hidden is None:

            hidden = self.config.getlist('tailbone', 'engines.trainwreck.hidden', 

                                         default=None)

            if hidden is not None:

                warnings.warn("[tailbone] 'engines.trainwreck.hidden' is a "

                              "deprecated setting, please use "

                              "[trainwreck.db] 'hide' instead",

                              DeprecationWarning)

        return hidden or []

    def engine_is_hidden(self, key):

        """

        Returns a boolean indicating if the given Trainwreck database

        engine is configured to be hidden from the user interface.

        """

        hidden = self.get_hidden_engine_keys()

        return key in hidden

    def get_oldest_transaction_date(self, session):

        """

        Query a Trainwreck database to determine the date of the

        "oldest" transaction it contains.

        :param session: SQLAlchemy session for a Trainwreck database.

        :return: A :class:`python:datetime.date` instance representing

           the oldest transaction date contained by the database.

        """

        trainwreck = self.config.get_trainwreck_model()

        txn = session.query(trainwreck.Transaction)\

                     .order_by(trainwreck.Transaction.end_time)\

                     .first()

        if txn:

            return self.app.localtime(txn.end_time, from_utc=True).date()

    def get_newest_transaction_date(self, session):

        """

        Query a Trainwreck database to determine the date of the

        "newest" transaction it contains.

        :param session: SQLAlchemy session for a Trainwreck database.

        :return: A :class:`python:datetime.date` instance representing

           the newest transaction date contained by the database.

        """

        trainwreck = self.config.get_trainwreck_model()

        txn = session.query(trainwreck.Transaction)\

                     .order_by(trainwreck.Transaction.end_time.desc())\

                     .first()

        if txn:

            return self.app.localtime(txn.end_time, from_utc=True).date()

    def test_get_trainwreck_handler(self):

        # first call gets the default handler

        trainwreck01 = self.handler.get_trainwreck_handler()

        self.assertIsNotNone(trainwreck01)

        # second call gets the same handler instance

        trainwreck02 = self.handler.get_trainwreck_handler()

        self.assertIs(trainwreck02, trainwreck01)

# -*- coding: utf-8; -*-

from __future__ import unicode_literals, absolute_import

import datetime

import warnings

from unittest import TestCase

import sqlalchemy as sa

from rattail.trainwreck import handler as mod

from rattail.config import make_config

from rattail.util import OrderedDict

from rattail.trainwreck.db import Session as TrainwreckSession

class TestTrainwreckHandler(TestCase):

    def setUp(self):

        self.config = self.make_config()

        self.handler = self.make_handler()

    def make_config(self):

        return make_config([], extend=False)

    def make_handler(self):

        return mod.TrainwreckHandler(self.config)

    def test_get_trainwreck_engines(self):

        # first let's configure 3 engines, 1 of which is hidden

        self.config.trainwreck_engines = OrderedDict([

            ('default', sa.create_engine('sqlite://')),

            ('2022', sa.create_engine('sqlite://')),

            ('2021', sa.create_engine('sqlite://')),

        ])

        self.config.setdefault('trainwreck.db', 'hide', '2022')

        # all 3 are returned by default

        engines = self.handler.get_trainwreck_engines()

        self.assertEqual(len(engines), 3)

        # but only 2 if we omit hidden

        engines = self.handler.get_trainwreck_engines(include_hidden=False)

        self.assertEqual(len(engines), 2)

        self.assertIn('default', engines)

        self.assertIn('2021', engines)

        self.assertNotIn('2022', engines)

    def test_get_hidden_engine_keys(self):

        # empty list returned by default

        result = self.handler.get_hidden_engine_keys()

        self.assertIsInstance(result, list)

        self.assertEqual(len(result), 0)

        # try the "legacy" setting first, to make testing simpler

        self.config.setdefault('tailbone', 'engines.trainwreck.hidden',

                               '2020, 2019, 2018')

        with warnings.catch_warnings(record=True):

            hidden = self.handler.get_hidden_engine_keys()

        self.assertEqual(hidden, ['2020', '2019', '2018'])

        # now try the "proper" setting

        self.config.setdefault('trainwreck.db', 'hide',

                               '2023, 2022, 2021')

        hidden = self.handler.get_hidden_engine_keys()

        self.assertEqual(hidden, ['2023', '2022', '2021'])

    def test_engine_is_hidden(self):

        # all engines are *not* hidden by default

        self.assertFalse(self.handler.engine_is_hidden('foobar'))

        # but any we explicitly hide should be reflected in call

        self.config.setdefault('trainwreck.db', 'hide',

                               '2023, 2022, 2021')

        self.assertTrue(self.handler.engine_is_hidden('2023'))

        self.assertTrue(self.handler.engine_is_hidden('2021'))

        self.assertFalse(self.handler.engine_is_hidden('2020'))

    def test_get_oldest_transaction_date(self):

        engine = sa.create_engine('sqlite://')

        self.config.setdefault('rattail', 'timezone.default',

                               'America/Chicago')

        self.config.setdefault('rattail.trainwreck', 'model',

                               'rattail.trainwreck.db.model.defaults')

        trainwreck = self.config.get_trainwreck_model()

        trainwreck.Base.metadata.create_all(bind=engine)

        session = TrainwreckSession(bind=engine)

        # empty db means oldest date is null

        result = self.handler.get_oldest_transaction_date(session)

        self.assertIsNone(result)

        # but if we insert a transaction, that date should be oldest

        dt = datetime.datetime(2022, 1, 1, 8)

        dt = self.handler.app.localtime(dt)

        txn = trainwreck.Transaction(end_time=dt)

        session.add(txn)

        result = self.handler.get_oldest_transaction_date(session)

        self.assertEqual(result, datetime.date(2022, 1, 1))

        # unless of course we add an older one..

        dt = datetime.datetime(2019, 6, 3, 12)

        dt = self.handler.app.localtime(dt)

        txn = trainwreck.Transaction(end_time=dt)

        session.add(txn)

        result = self.handler.get_oldest_transaction_date(session)

        self.assertEqual(result, datetime.date(2019, 6, 3))

    def test_get_newest_transaction_date(self):

        engine = sa.create_engine('sqlite://')

        self.config.setdefault('rattail', 'timezone.default',

                               'America/Chicago')

        self.config.setdefault('rattail.trainwreck', 'model',

                               'rattail.trainwreck.db.model.defaults')

        trainwreck = self.config.get_trainwreck_model()

        trainwreck.Base.metadata.create_all(bind=engine)

        session = TrainwreckSession(bind=engine)

        # empty db means newest date is null

        result = self.handler.get_newest_transaction_date(session)

        self.assertIsNone(result)

        # but if we insert a transaction, that date should be newest

        dt = datetime.datetime(2019, 6, 3, 12)

        dt = self.handler.app.localtime(dt)

        txn = trainwreck.Transaction(end_time=dt)

        session.add(txn)

        result = self.handler.get_newest_transaction_date(session)

        self.assertEqual(result, datetime.date(2019, 6, 3))

        # unless of course we add an newer one..

        dt = datetime.datetime(2022, 1, 1, 8)

        dt = self.handler.app.localtime(dt)

        txn = trainwreck.Transaction(end_time=dt)

        session.add(txn)

        result = self.handler.get_newest_transaction_date(session)

        self.assertEqual(result, datetime.date(2022, 1, 1))