<commit_before>#!/usr/bin/env python

__author__ = '<NAME>, <NAME>'
__license__ = 'Apache 2.0'

from pyon.core.bootstrap import obj_registry, IonObject
from pyon.core.exception import NotFound
from pyon.datastore.datastore import DataStore
from pyon.datastore.mockdb.mockdb_datastore import MockDB_DataStore
from pyon.datastore.couchdb.couchdb_datastore import CouchDB_DataStore
from pyon.util.int_test import IonIntegrationTestCase
from pyon.public import RT, AT, LCS
from nose.plugins.attrib import attr
from unittest import SkipTest
import socket

OWNER_OF = "XOWNER_OF"
HAS_A = "XHAS_A"
BASED_ON = "XBASED_ON"

@attr('UNIT')
class Test_DataStores(IonIntegrationTestCase):

    def test_non_persistent(self):
        self._do_test(MockDB_DataStore(datastore_name='my_ds'))
        
        self._do_test_views(MockDB_DataStore(datastore_name='my_ds'))

    def test_persistent(self):
        try:
            self._do_test(CouchDB_DataStore(datastore_name='my_ds'))

            self._do_test_views(CouchDB_DataStore(datastore_name='my_ds'), is_persistent=True)
        except socket.error:
            raise SkipTest('Failed to connect to CouchDB')

    def _do_test(self, data_store):
        self.data_store = data_store
        self.resources = {}
        # Just in case previous run failed without cleaning up,
        # delete data store
        try:
            data_store.delete_datastore()
        except NotFound:
            pass

        # Create should succeed and not throw error<commit_msg>Fixed persistent datastore test<commit_after>#!/usr/bin/env python

__author__ = '<NAME>, <NAME>'
__license__ = 'Apache 2.0'

from pyon.core.bootstrap import obj_registry, IonObject
from pyon.core.exception import NotFound
from pyon.datastore.datastore import DataStore
from pyon.datastore.mockdb.mockdb_datastore import MockDB_DataStore
from pyon.datastore.couchdb.couchdb_datastore import CouchDB_DataStore
from pyon.util.int_test import IonIntegrationTestCase
from pyon.public import RT, AT, LCS
from nose.plugins.attrib import attr
from unittest import SkipTest
import socket

OWNER_OF = "XOWNER_OF"
HAS_A = "XHAS_A"
BASED_ON = "XBASED_ON"

@attr('UNIT')
class Test_DataStores(IonIntegrationTestCase):

    def test_non_persistent(self):
        self._do_test(MockDB_DataStore(datastore_name='my_ds'))
        
        self._do_test_views(MockDB_DataStore(datastore_name='my_ds'))

    def test_persistent(self):
        import socket
        try:
            self._do_test(CouchDB_DataStore(datastore_name='my_ds'))

            self._do_test_views(CouchDB_DataStore(datastore_name='my_ds'), is_persistent=True)
        except socket.error:
            raise SkipTest('Failed to connect to CouchDB')

    def _do_test(self, data_store):
        self.data_store = data_store
        self.resources = {}
        # Just in case previous run failed without cleaning up,
        # delete data store
        try:
            data_store.delete_datastore()
        except NotFound:
            pass

        # Create should succeed and not throw error
{'fraction_non_alphanumeric': 0.08049738219895287, 'fraction_numerical': 0.0013089005235602095, 'mean_word_length': 3.631818181818182, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3209549', 'n_tokens_mistral': 1002, 'n_tokens_neox': 977, 'n_words': 216}