``rattail.commands`` -- Commands

"""

import sys

import platform

import edbob

from edbob import commands

from edbob.commands import Subcommand

from ._version import __version__

from .db import model

class Command(commands.Command):

    """

    The primary command for Rattail.

    """

    name = 'rattail'

    version = __version__

    description = "Retail Software Framework"

    long_description = """

Rattail is a retail software framework.

    def add_parser_args(self, parser):

        parser.add_argument(

            '--output', '-o', metavar='FILE',

            help="Optional path to output file.  If none is specified, "

            "data will be written to standard output.")

        parser.add_argument(

            'model', help="Model whose data will be dumped.")

    def run(self, args):

        from .db import get_session_class

        from .db.dump import dump_data

        if hasattr(model, args.model):

            cls = getattr(model, args.model)

        else:

            sys.stderr.write("Unknown model: {0}\n".format(args.model))

            sys.exit(1)

        progress = None

        if self.show_progress:

            progress = Progress

        if args.output:

        print("  {0}".format(obfuscate_url_pw(engine.url)))

class LoadHostDataCommand(Subcommand):

    """

    Loads data from the Rattail host database, if one is configured.

    """

    name = 'load-host-data'

    description = "Load data from host database"

    def run(self, args):

        proc = load.LoadProcessor()

class MakeConfigCommand(Subcommand):

    """

    Creates a sample configuration file.

    """

    name = 'make-config'

    description = "Create a configuration file"

    def add_parser_args(self, parser):

        parser.add_argument('path', default='rattail.conf', metavar='PATH',

import time

import logging

if sys.platform == 'win32': # pragma no cover

    import win32api

import sqlalchemy.exc

from sqlalchemy.orm import sessionmaker, class_mapper

from sqlalchemy.exc import OperationalError

import edbob

log = logging.getLogger(__name__)

def get_sync_engines():

    keys = edbob.config.get('rattail.db', 'syncs')

    if not keys:

        return None

    for key in keys.split(','):

        key = key.strip()

class Synchronizer(object):

    """

    Default implementation of database synchronization logic.  Subclass this if

    you have special processing needs.

    """

    # This defines the `model` module which will be used to obtain references

    # to model classes (`Product` etc.).  If you need to synchronize custom

    # model classes of which Rattail is not aware, you must override this.

    # Note that the module you specify must be a superset of Rattail.

#  more details.

#

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

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

#

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

"""

``rattail.db.sync.linux`` -- Database Synchronization for Linux

"""

import edbob

from ...daemon import Daemon

from . import get_sync_engines, synchronize_changes

class SyncDaemon(Daemon):

    def run(self):

        remote_engines = get_sync_engines()

        if remote_engines:

def get_daemon(pidfile=None):

    """

    Get a :class:`SyncDaemon` instance.

    """

    if pidfile is None:

        pidfile = edbob.config.get('rattail.db', 'sync.pid_path',

                                   default='/var/run/rattail/dbsync.pid')

    return SyncDaemon(pidfile)

#

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

"""

``rattail.db.sync.win32`` -- Database Synchronization for Windows

"""

import sys

import logging

import threading

import edbob

log = logging.getLogger(__name__)

class DatabaseSynchronizerService(Service):

    """

    Implements database synchronization as a Windows service.

    """

    _svc_name_ = 'RattailDatabaseSynchronizer'

    _svc_display_name_ = "Rattail : Database Synchronization Service"

                         "database(s).")

    appname = 'rattail'

    def Initialize(self):

        """

        Service initialization.

        """

        if not Service.Initialize(self):

            return False

        remote_engines = get_sync_engines()

        if not remote_engines:

            return False

        thread = threading.Thread(target=synchronize_changes,

        thread.daemon = True

        thread.start()

        return True

if __name__ == '__main__':

    if sys.platform == 'win32':

        import win32serviceutil

        win32serviceutil.HandleCommandLine(DatabaseSynchronizerService)

        session.flush()

        assignment = session.query(model.CustomerGroupAssignment).one()

        self.assertEqual(assignment.customer_uuid, customer.uuid)

        self.assertEqual(assignment.group_uuid, group.uuid)

        synchronizer.delete_CustomerGroup(session, group)

        self.assertEqual(session.query(model.CustomerGroupAssignment).count(), 0)

        session.rollback()

        session.close()

class ModuleTests(TestCase):

        # nothing configured

        edbob.config.get.return_value = None

        self.assertIsNone(sync.get_sync_engines())

        # fake config with 2 out of 3 engines synced

            'one': 'first',

            'two': 'second',

            'three': 'third',

            }

        edbob.config.get.return_value = 'one, two'

        engines = sync.get_sync_engines()

        self.assertEqual(engines, {'one': 'first', 'two': 'second'})

    @patch.multiple('rattail.db.sync', edbob=DEFAULT, Synchronizer=DEFAULT)

    def test_synchronize_changes(self, edbob, Synchronizer):

        local_engine = Mock()

from unittest import TestCase

from mock import patch, DEFAULT

from rattail.db.sync import linux

class SyncDaemonTests(TestCase):

    @patch.multiple('rattail.db.sync.linux',

                    get_sync_engines=DEFAULT,

                    synchronize_changes=DEFAULT)

        daemon = linux.SyncDaemon('/tmp/rattail_dbsync.pid')

        # no remote engines configured

        get_sync_engines.return_value = None

        daemon.run()

        get_sync_engines.assert_called_once_with()

        self.assertFalse(synchronize_changes.called)

        # with remote engines configured

        get_sync_engines.return_value = 'fake_remotes'

        daemon.run()

class ModuleTests(TestCase):

    @patch.multiple('rattail.db.sync.linux', edbob=DEFAULT, SyncDaemon=DEFAULT)

    def test_get_daemon(self, edbob, SyncDaemon):

        # pid file provided

        linux.get_daemon('some_pidfile')

        self.assertFalse(edbob.config.get.called)

        SyncDaemon.assert_called_once_with('some_pidfile')