<commit_before># -*- coding: utf-8 -*-
# vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright (c) 2010-2011, GEM Foundation.
#
# OpenQuake is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License version 3
# only, as published by the Free Software Foundation.
#
# OpenQuake 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 Lesser General Public License version 3 for more details
# (a copy is included in the LICENSE file that accompanied this code).
#
# You should have received a copy of the GNU Lesser General Public License
# version 3 along with OpenQuake.  If not, see
# <http://www.gnu.org/licenses/lgpl-3.0.txt> for a copy of the LGPLv3 License.


import json
import mock
import os

import unittest

from openquake import java
from openquake import logs
from openquake import kvs
from openquake import settings
from tests.utils import helpers


from openquake.kvs import tokens


LOG = logs.LOG

TEST_FILE = "nrml_test_result.xml"

EMPTY_MODEL = '{"modelName":"","hcRepList":[],"endBranchLabels":[]}'

TEST_KEY = "a key for testing purposes"


def read_one_line(path):
    """
    Read and return a single line from the given file.

    :param path: path to a (text) file
    :type path: str

    :returns: first line from the file
    """
    return open(path, 'r').readline().strip('\n')


ONE_CURVE_MODEL = read_one_line(helpers.get_data_path('one-curve-model.json'))


class KVSTestCase(unittest.TestCase):
    """
    Tests for various KVS storage operations.
    """

    def setUp(self):
        # starting the jvm...
        print "About to start the jvm..."
        jpype = java.jvm()
        java_class = jpype.JClass("org.gem.engine.hazard.redis.Cache")
        print "Not dead yet, and found the class..."
        self.java_client = java_class(settings.KVS_HOST, settings.KVS_PORT)

        self.python_client = kvs.get_client(binary=False)

        self._delete_test_file()

    def tearDown(self):
        self._delete_test_file()
        self.python_client.flushdb()

    @staticmethod
    def _delete_test_file():
        try:
            os.remove(os.path.join(helpers.DATA_DIR, TEST_FILE))
        except OSError:
            pass

    def test_can_wrap_the_java_client(self):
        self.java_client.set("KEY", "VALUE")
        self.assertEqual("VALUE", self.java_client.get("KEY"))

    def test_can_write_in_java_and_read_in_python(self):
        self.java_client.set("KEY", "VALUE")
        self.assertEqual("VALUE", self.python_client.get("KEY"))

    def test_can_write_in_python_and_read_in_java(self):
        self.python_client.set("KEY", "VALUE")
        self.assertEqual("VALUE", self.java_client.get("KEY"))

    def test_get_list_json_decoded(self):
        data = [{u'1': u'one'}, {u'2': u'two'}, {u'3': u'three'}]

        for item in data:
            kvs.get_client().rpush(TEST_KEY, json.dumps(item))

        self.assertEqual(kvs.get_list_json_decoded(TEST_KEY), data)


class TokensTestCase(unittest.TestCase):
    """
    Tests for functions related to generation/allocation of KVS keys.
    """

    def setUp(self):
        self.job_id = 123456
        self.product = "TestProduct"<commit_msg>Swapped assertEqual args<commit_after># -*- coding: utf-8 -*-
# vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright (c) 2010-2011, GEM Foundation.
#
# OpenQuake is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License version 3
# only, as published by the Free Software Foundation.
#
# OpenQuake 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 Lesser General Public License version 3 for more details
# (a copy is included in the LICENSE file that accompanied this code).
#
# You should have received a copy of the GNU Lesser General Public License
# version 3 along with OpenQuake.  If not, see
# <http://www.gnu.org/licenses/lgpl-3.0.txt> for a copy of the LGPLv3 License.


import json
import mock
import os

import unittest

from openquake import java
from openquake import logs
from openquake import kvs
from openquake import settings
from tests.utils import helpers


from openquake.kvs import tokens


LOG = logs.LOG

TEST_FILE = "nrml_test_result.xml"

EMPTY_MODEL = '{"modelName":"","hcRepList":[],"endBranchLabels":[]}'

TEST_KEY = "a key for testing purposes"


def read_one_line(path):
    """
    Read and return a single line from the given file.

    :param path: path to a (text) file
    :type path: str

    :returns: first line from the file
    """
    return open(path, 'r').readline().strip('\n')


ONE_CURVE_MODEL = read_one_line(helpers.get_data_path('one-curve-model.json'))


class KVSTestCase(unittest.TestCase):
    """
    Tests for various KVS storage operations.
    """

    def setUp(self):
        # starting the jvm...
        print "About to start the jvm..."
        jpype = java.jvm()
        java_class = jpype.JClass("org.gem.engine.hazard.redis.Cache")
        print "Not dead yet, and found the class..."
        self.java_client = java_class(settings.KVS_HOST, settings.KVS_PORT)

        self.python_client = kvs.get_client(binary=False)

        self._delete_test_file()

    def tearDown(self):
        self._delete_test_file()
        self.python_client.flushdb()

    @staticmethod
    def _delete_test_file():
        try:
            os.remove(os.path.join(helpers.DATA_DIR, TEST_FILE))
        except OSError:
            pass

    def test_can_wrap_the_java_client(self):
        self.java_client.set("KEY", "VALUE")
        self.assertEqual("VALUE", self.java_client.get("KEY"))

    def test_can_write_in_java_and_read_in_python(self):
        self.java_client.set("KEY", "VALUE")
        self.assertEqual("VALUE", self.python_client.get("KEY"))

    def test_can_write_in_python_and_read_in_java(self):
        self.python_client.set("KEY", "VALUE")
        self.assertEqual("VALUE", self.java_client.get("KEY"))

    def test_get_list_json_decoded(self):
        data = [{u'1': u'one'}, {u'2': u'two'}, {u'3': u'three'}]

        for item in data:
            kvs.get_client().rpush(TEST_KEY, json.dumps(item))

        self.assertEqual(data, kvs.get_list_json_decoded(TEST_KEY))


class TokensTestCase(unittest.TestCase):
    """
    Tests for functions related to generation/allocation of KVS keys.
    """

    def setUp(self):
        self.job_id = 123456
        self.product = "TestProduct"
{'fraction_non_alphanumeric': 0.10840674929072719, 'fraction_numerical': 0.008063311930715246, 'mean_word_length': 3.324080051646223, 'pattern_counts': {'":': 6, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 2}, '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': '882164', 'n_tokens_mistral': 2153, 'n_tokens_neox': 2088, 'n_words': 607}