Peler

: / opt / cpanel-ccs / txdav / base / datastore /
Filename : suboracle.py
back
# -*- test-case-name: txdav.base.datastore.test.test_subpostgres -*-
# #
# Copyright (c) 2010-2017 Apple Inc. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# #

"""
Stub service for Oracle.
"""

from twext.python.log import Logger

from txdav.base.datastore.dbapiclient import DBAPIConnector

from twisted.application.service import MultiService
from twisted.internet.defer import inlineCallbacks

log = Logger()


class OracleService(MultiService):
    """
    A service for connecting to a locally running Oracle DB VM. This is only used
    during unit tests.
    """

    def __init__(
        self, dataStoreDirectory, subServiceFactory,
        dsnUser=None,
        testMode=False,
        reactor=None,
    ):
        """
        Initialize a L{OracleService} pointed at a data store directory.

        @param dataStoreDirectory: the directory to
        @type dataStoreDirectory: L{twext.python.filepath.CachingFilePath}

        @param subServiceFactory: a 1-arg callable that will be called with a
            1-arg callable which returns a DB-API cursor.
        @type subServiceFactory: C{callable}
        """

        MultiService.__init__(self)
        self.subServiceFactory = subServiceFactory
        self.dataStoreDirectory = dataStoreDirectory
        self.workingDir = self.dataStoreDirectory.child("working")

        self.dsnUser = dsnUser
        self.testMode = testMode

        self._reactor = reactor

    @property
    def reactor(self):
        if self._reactor is None:
            from twisted.internet import reactor
            self._reactor = reactor
        return self._reactor

    def _connectorFor(self):
        kwargs = {
            "endpoint": "tcp:192.168.56.101:1521",
            "database": "orcl",
            "user": self.dsnUser if self.dsnUser else "hr",
            "password": "oracle",
        }

        return DBAPIConnector.connectorFor("oracle", **kwargs)

    def produceConnection(self, label="<unlabeled>"):
        """
        Produce a DB-API 2.0 connection pointed at this database.
        """
        return self._connectorFor().connect(label)

    def pauseMonitor(self):
        """
        Pause monitoring.
        """
        pass

    def unpauseMonitor(self):
        """
        Unpause monitoring.

        @see: L{pauseMonitor}
        """
        pass

    def startService(self):
        MultiService.startService(self)

        if not self.dataStoreDirectory.isdir():
            log.info("Creating {dir}", dir=self.dataStoreDirectory.path)
            self.dataStoreDirectory.createDirectory()

        if not self.workingDir.isdir():
            log.info("Creating {dir}", dir=self.workingDir.path)
            self.workingDir.createDirectory()

        self.subServiceFactory(
            self.produceConnection, self
        ).setServiceParent(self)

    def hardStop(self):
        """
        Stop quickly by sending it SIGQUIT
        """
        pass


@inlineCallbacks
def cleanDatabase(txn):
    tables = yield txn.execSQL("select table_name from user_tables")
    for table in tables:
        yield txn.execSQL("drop table {} cascade constraints purge".format(table[0]))
    yield txn.execSQL("purge recyclebin")

    sequences = yield txn.execSQL("select sequence_name from user_sequences")
    for sequence in sequences:
        yield txn.execSQL("drop sequence {}".format(sequence[0]))

    indexes = yield txn.execSQL("select index_name from user_indexes")
    for index in indexes:
        yield txn.execSQL("drop index {}".format(index[0]))

    constraints = yield txn.execSQL("select constraint_name from user_constraints")
    for constraint in constraints:
        yield txn.execSQL("drop constraint '{}'".format(constraint[0]))