<commit_before># Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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.
import httplib
import sys
import unittest
import urlparse

try:
    import json
except ImportError:
    import simplejson as json

from libcloud.types import LibcloudError, InvalidCredsError
from libcloud.drivers.gogrid import GoGridNodeDriver
from libcloud.base import Node, NodeImage, NodeSize, NodeLocation

from test import MockHttp, TestCaseMixin
from test.file_fixtures import FileFixtures


class GoGridTests(unittest.TestCase, TestCaseMixin):

    def setUp(self):
        GoGridNodeDriver.connectionCls.conn_classes = (None, GoGridMockHttp)
        GoGridMockHttp.type = None
        self.driver = GoGridNodeDriver("foo", "bar")

    def test_create_node(self):
        image = NodeImage(1531, None, self.driver)
        size = NodeSize('512Mb', None, None, None, None, None, driver=self.driver)

        node = self.driver.create_node(name='test1', image=image, size=size)
        self.assertEqual(node.name, 'test1')
        self.assertTrue(node.id is not None)
        self.assertEqual(node.extra['password'], '<PASSWORD>')

    def test_list_nodes(self):
        node = self.driver.list_nodes()[0]

        self.assertEqual(node.id, '90967')
        self.assertEqual(node.extra['password'], '<PASSWORD>')
        self.assertEqual(node.extra['isSandbox'], False)

    def test_reboot_node(self):
        node = Node(90967, None, None, None, None, self.driver)
        ret = self.driver.reboot_node(node)
        self.assertTrue(ret)

    def test_destroy_node(self):
        node = Node(90967, None, None, None, None, self.driver)
        ret = self.driver.destroy_node(node)
        self.assertTrue(ret)

    def test_list_images(self):
        images = self.driver.list_images()
        image = images[0]
        self.assertEqual(len(images), 4)
        self.assertEqual(image.name, 'CentOS 5.3 (32-bit) w/ None')
        self.assertEqual(image.id, '1531')

    def test_malformed_reply(self):
        GoGridMockHttp.type = 'FAIL'
        try:
            images = self.driver.list_images()
        except LibcloudError, e:
            self.assertTrue(isinstance(e, LibcloudError))
        else:
            self.fail("test should have thrown")

    def test_invalid_creds(self):
        GoGridMockHttp.type = 'FAIL'
        try:
            nodes = self.driver.list_nodes()
        except InvalidCredsError, e:
            self.assertTrue(e.driver is not None)
            self.assertEqual(e.driver.name, self.driver.name)
        else:
            self.fail("test should have thrown")

    def test_node_creation_without_free_public_ips(self):
        GoGridMockHttp.type = 'NOPUBIPS'
        try:
            image = NodeImage(1531, None, self.driver)
            size = NodeSize('512Mb', None, None, None, None, None, driver=self.driver)

            node = self.driver.create_node(name='test1', image=image, size=size)
        except LibcloudError, e:
            self.assertTrue(isinstance(e, LibcloudError))
            self.assertTrue(e.driver is not None)
            self.assertEqual(e.driver.name, self.driver.name)
        else:
            self.fail("test should have thrown")

    def test_list_locations(self):
        locations = self.driver.list_locations()
        location_names = [location.name for location in locations]

        self.assertEqual(len(locations), 2)
        for i in 0, 1:
            self.assertTrue(isinstance(locations[i], NodeLocation))
        self.assertTrue("US-West-1" in location_names)
        self.assertTrue("US-East-1" in location_names)

    def test_ex_save_image(self):
        node = self.driver.list_nodes()[0]
        image = self.driver.ex_save_image(node, "testimage")
        self.assertEqual(image.name, "testimage")

class GoGridMockHttp(MockHttp):

    fixtures = FileFixtures('gogrid')

    def _api_grid_image_list(self, method, url, body, headers):
        body = self.fixtures.load('image_list.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    def _api_grid_image_list_FAIL(self, method, url, body, headers):
        body = "<h3>some non valid json here</h3>"
        return (httplib.SERVICE_UNAVAILABLE, body, {},
                httplib.responses[httplib.SERVICE_UNAVAILABLE])

    def _api_grid_server_list(self, method, url, body, headers):
        body = self.fixtures.load('server_list.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    _api_grid_server_list_NOPUBIPS = _api_grid_server_list

    def _api_grid_server_list_FAIL(self, method, url, body, headers):
        return (httplib.FORBIDDEN, "123", {}, httplib.responses[httplib.FORBIDDEN])

    def _api_grid_ip_list(self, method, url, body, headers):
        body = self.fixtures.load('ip_list.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    def _api_grid_ip_list_NOPUBIPS(self, method, url, body, headers):
        body = self.fixtures.load('ip_list_empty.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    def _api_grid_server_power(self, method, url, body, headers):
        body = self.fixtures.load('server_power.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    def _api_grid_server_add(self, method, url, body, headers):
        body = self.fixtures.load('server_add.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    _api_grid_server_add_NOPUBIPS = _api_grid_server_add

    def _api_grid_server_delete(self, method, url, body, headers):
        body = self.fixtures.load('server_delete.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    def _api_support_password_list(self, method, url, body, headers):
        body = self.fixtures.load('password_list.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    _api_support_password_list_NOPUBIPS = _api_support_password_list

    def _api_grid_image_save(self, method, url, body, headers):
        body = self.fixtures.load('image_save.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    def _api_common_lookup_list(self, method, url, body, headers):
        _valid_lookups = ("ip.datacenter",)

        lookup = urlparse.parse_qs(
                    urlparse.urlparse(url).query)["lookup"][0]
        if lookup in _valid_lookups:
            fixture_path = "lookup_list_%s.json" % \
                    (lookup.replace(".", "_"))
        else:
            raise NotImplementedError
        body = self.fixtures.load(fixture_path)
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

if __name__ == '__main__':
    sys.exit(unittest.main())
<commit_msg>Fallback for urlparse.parse_qs to use (now deprecated) cgi.parse_qs in < 2.6<commit_after># Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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.
import httplib
import sys
import unittest
import urlparse

try:
    import json
except ImportError:
    import simplejson as json

from libcloud.types import LibcloudError, InvalidCredsError
from libcloud.drivers.gogrid import GoGridNodeDriver
from libcloud.base import Node, NodeImage, NodeSize, NodeLocation

from test import MockHttp, TestCaseMixin
from test.file_fixtures import FileFixtures


class GoGridTests(unittest.TestCase, TestCaseMixin):

    def setUp(self):
        GoGridNodeDriver.connectionCls.conn_classes = (None, GoGridMockHttp)
        GoGridMockHttp.type = None
        self.driver = GoGridNodeDriver("foo", "bar")

    def test_create_node(self):
        image = NodeImage(1531, None, self.driver)
        size = NodeSize('512Mb', None, None, None, None, None, driver=self.driver)

        node = self.driver.create_node(name='test1', image=image, size=size)
        self.assertEqual(node.name, 'test1')
        self.assertTrue(node.id is not None)
        self.assertEqual(node.extra['password'], '<PASSWORD>')

    def test_list_nodes(self):
        node = self.driver.list_nodes()[0]

        self.assertEqual(node.id, '90967')
        self.assertEqual(node.extra['password'], '<PASSWORD>')
        self.assertEqual(node.extra['isSandbox'], False)

    def test_reboot_node(self):
        node = Node(90967, None, None, None, None, self.driver)
        ret = self.driver.reboot_node(node)
        self.assertTrue(ret)

    def test_destroy_node(self):
        node = Node(90967, None, None, None, None, self.driver)
        ret = self.driver.destroy_node(node)
        self.assertTrue(ret)

    def test_list_images(self):
        images = self.driver.list_images()
        image = images[0]
        self.assertEqual(len(images), 4)
        self.assertEqual(image.name, 'CentOS 5.3 (32-bit) w/ None')
        self.assertEqual(image.id, '1531')

    def test_malformed_reply(self):
        GoGridMockHttp.type = 'FAIL'
        try:
            images = self.driver.list_images()
        except LibcloudError, e:
            self.assertTrue(isinstance(e, LibcloudError))
        else:
            self.fail("test should have thrown")

    def test_invalid_creds(self):
        GoGridMockHttp.type = 'FAIL'
        try:
            nodes = self.driver.list_nodes()
        except InvalidCredsError, e:
            self.assertTrue(e.driver is not None)
            self.assertEqual(e.driver.name, self.driver.name)
        else:
            self.fail("test should have thrown")

    def test_node_creation_without_free_public_ips(self):
        GoGridMockHttp.type = 'NOPUBIPS'
        try:
            image = NodeImage(1531, None, self.driver)
            size = NodeSize('512Mb', None, None, None, None, None, driver=self.driver)

            node = self.driver.create_node(name='test1', image=image, size=size)
        except LibcloudError, e:
            self.assertTrue(isinstance(e, LibcloudError))
            self.assertTrue(e.driver is not None)
            self.assertEqual(e.driver.name, self.driver.name)
        else:
            self.fail("test should have thrown")

    def test_list_locations(self):
        locations = self.driver.list_locations()
        location_names = [location.name for location in locations]

        self.assertEqual(len(locations), 2)
        for i in 0, 1:
            self.assertTrue(isinstance(locations[i], NodeLocation))
        self.assertTrue("US-West-1" in location_names)
        self.assertTrue("US-East-1" in location_names)

    def test_ex_save_image(self):
        node = self.driver.list_nodes()[0]
        image = self.driver.ex_save_image(node, "testimage")
        self.assertEqual(image.name, "testimage")

class GoGridMockHttp(MockHttp):

    fixtures = FileFixtures('gogrid')

    def _api_grid_image_list(self, method, url, body, headers):
        body = self.fixtures.load('image_list.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    def _api_grid_image_list_FAIL(self, method, url, body, headers):
        body = "<h3>some non valid json here</h3>"
        return (httplib.SERVICE_UNAVAILABLE, body, {},
                httplib.responses[httplib.SERVICE_UNAVAILABLE])

    def _api_grid_server_list(self, method, url, body, headers):
        body = self.fixtures.load('server_list.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    _api_grid_server_list_NOPUBIPS = _api_grid_server_list

    def _api_grid_server_list_FAIL(self, method, url, body, headers):
        return (httplib.FORBIDDEN, "123", {}, httplib.responses[httplib.FORBIDDEN])

    def _api_grid_ip_list(self, method, url, body, headers):
        body = self.fixtures.load('ip_list.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    def _api_grid_ip_list_NOPUBIPS(self, method, url, body, headers):
        body = self.fixtures.load('ip_list_empty.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    def _api_grid_server_power(self, method, url, body, headers):
        body = self.fixtures.load('server_power.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    def _api_grid_server_add(self, method, url, body, headers):
        body = self.fixtures.load('server_add.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    _api_grid_server_add_NOPUBIPS = _api_grid_server_add

    def _api_grid_server_delete(self, method, url, body, headers):
        body = self.fixtures.load('server_delete.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    def _api_support_password_list(self, method, url, body, headers):
        body = self.fixtures.load('password_list.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    _api_support_password_list_NOPUBIPS = _api_support_password_list

    def _api_grid_image_save(self, method, url, body, headers):
        body = self.fixtures.load('image_save.json')
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

    def _api_common_lookup_list(self, method, url, body, headers):
        _valid_lookups = ("ip.datacenter",)

        try:
            from urlparse import parse_qs
        except ImportError:
            from cgi import parse_qs

        lookup = parse_qs(urlparse.urlparse(url).query)["lookup"][0]
        if lookup in _valid_lookups:
            fixture_path = "lookup_list_%s.json" % \
                    (lookup.replace(".", "_"))
        else:
            raise NotImplementedError
        body = self.fixtures.load(fixture_path)
        return (httplib.OK, body, {}, httplib.responses[httplib.OK])

if __name__ == '__main__':
    sys.exit(unittest.main())

{'fraction_non_alphanumeric': 0.11026084642001598, 'fraction_numerical': 0.007985094490284802, 'mean_word_length': 3.1608527131782944, 'pattern_counts': {'":': 0, '<': 12, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '3471648', 'n_tokens_mistral': 4753, 'n_tokens_neox': 4638, 'n_words': 1075}