<commit_before># Copyright (C) 2009 <NAME> <<EMAIL>>
# Licenced under the txaws licence available at /LICENSE in the txaws source.

import os

from txaws.credentials import AWSCredentials
from txaws.ec2.client import EC2Client
from txaws.service import (AWSServiceEndpoint, AWSServiceRegion,
                           EC2_ENDPOINT_US, EC2_ENDPOINT_EU, REGION_EU)
from txaws.tests import TXAWSTestCase


class AWSServiceEndpointTestCase(TXAWSTestCase):

    def setUp(self):
        self.endpoint = AWSServiceEndpoint(uri="http://my.service/da_endpoint")

    def test_simple_creation(self):
        endpoint = AWSServiceEndpoint()
        self.assertEquals(endpoint.scheme, "http")
        self.assertEquals(endpoint.host, "")
        self.assertEquals(endpoint.port, 80)
        self.assertEquals(endpoint.path, "/")
        self.assertEquals(endpoint.method, "GET")

    def test_parse_uri(self):
        self.assertEquals(self.endpoint.scheme, "http")
        self.assertEquals(self.endpoint.host, "my.service")
        self.assertEquals(self.endpoint.port, 80)
        self.assertEquals(self.endpoint.path, "/da_endpoint")

    def test_parse_uri_https_and_custom_port(self):
        endpoint = AWSServiceEndpoint(uri="https://my.service:8080/endpoint")
        self.assertEquals(endpoint.scheme, "https")
        self.assertEquals(endpoint.host, "my.service")
        self.assertEquals(endpoint.port, 8080)
        self.assertEquals(endpoint.path, "/endpoint")

    def test_custom_method(self):
        endpoint = AWSServiceEndpoint(uri="http://service/endpoint",
                                      method="PUT")
        self.assertEquals(endpoint.method, "PUT")

    def test_get_uri(self):
        uri = self.endpoint.get_uri()
        self.assertEquals(uri, "http://my.service/da_endpoint")

    def test_get_uri_custom_port(self):
        uri = "https://my.service:8080/endpoint"
        endpoint = AWSServiceEndpoint(uri=uri)
        new_uri = endpoint.get_uri()
        self.assertEquals(new_uri, uri)

    def test_set_path(self):
        original_path = self.endpoint.path
        self.endpoint.set_path("/newpath")
        self.assertEquals(
            self.endpoint.get_uri(),
            "http://my.service/newpath")


class AWSServiceRegionTestCase(TXAWSTestCase):

    def setUp(self):
        self.creds = AWSCredentials("foo", "bar")
        self.region = AWSServiceRegion(creds=self.creds)

    def test_simple_creation(self):
        self.assertEquals(self.creds, self.region.creds)
        self.assertEquals(self.region._clients, {})
        self.assertEquals(self.region.ec2_endpoint.get_uri(), EC2_ENDPOINT_US)

    def test_creation_with_keys(self):
        region = AWSServiceRegion(access_key="baz", secret_key="quux")
        self.assertEquals(region.creds.access_key, "baz")
        self.assertEquals(region.creds.secret_key, "quux")

    def test_creation_with_keys_and_creds(self):
        region = AWSServiceRegion(self.creds, access_key="baz",
                                  secret_key="quux")
        self.assertEquals(self.creds.access_key, "foo")
        self.assertEquals(self.creds.secret_key, "bar")

    def test_creation_with_uri(self):
        region = AWSServiceRegion(creds=self.creds, uri="http://foo/bar")
        self.assertEquals(region.ec2_endpoint.get_uri(), "http://foo/bar")

    def test_creation_with_uri_and_region(self):
        region = AWSServiceRegion(creds=self.creds, region=REGION_EU,
                                  uri="http://foo/bar")
        self.assertEquals(region.ec2_endpoint.get_uri(), "http://foo/bar")

    def test_creation_with_region_override(self):
        region = AWSServiceRegion(creds=self.creds, region=REGION_EU)
        self.assertEquals(region.ec2_endpoint.get_uri(), EC2_ENDPOINT_EU)

    def test_get_client_with_empty_cache(self):
        key = str(EC2Client) + str(self.creds) + str(self.region.ec2_endpoint)
        original_client = self.region._clients.get(key)
        new_client = self.region.get_client(
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint)
        self.assertEquals(original_client, None)
        self.assertTrue(isinstance(new_client, EC2Client))
        self.assertNotEquals(original_client, new_client)

    def test_get_client_from_cache(self):
        client1 = self.region.get_client(
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint)
        client2 = self.region.get_client(
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint)
        self.assertTrue(isinstance(client1, EC2Client))
        self.assertTrue(isinstance(client2, EC2Client))
        self.assertEquals(client1, client2)

    def test_get_client_from_cache_with_purge(self):
        client1 = self.region.get_client(
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint,
            purge_cache=True)
        client2 = self.region.get_client(
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint,
            purge_cache=True)
        self.assertTrue(isinstance(client1, EC2Client))
        self.assertTrue(isinstance(client2, EC2Client))
        self.assertNotEquals(client1, client2)

    def test_get_ec2_client_from_cache(self):
        client1 = self.region.get_ec2_client(self.creds)
        client2 = self.region.get_ec2_client(self.creds)
        self.assertEquals(self.creds, self.region.creds)
        self.assertTrue(isinstance(client1, EC2Client))
        self.assertTrue(isinstance(client2, EC2Client))
        self.assertEquals(client1, client2)


    def test_get_s3_client(self):
        self.assertRaises(NotImplementedError, self.region.get_s3_client)

    def test_get_simpledb_client(self):
        self.assertRaises(NotImplementedError, self.region.get_simpledb_client)

    def test_get_sqs_client(self):
        self.assertRaises(NotImplementedError, self.region.get_sqs_client)
<commit_msg>Swapped the ordering of an import to be in alphabetical order.<commit_after># Copyright (C) 2009 <NAME> <<EMAIL>>
# Licenced under the txaws licence available at /LICENSE in the txaws source.

import os

from txaws.credentials import AWSCredentials
from txaws.ec2.client import EC2Client
from txaws.service import (AWSServiceEndpoint, AWSServiceRegion,
                           EC2_ENDPOINT_EU, EC2_ENDPOINT_US, REGION_EU)
from txaws.tests import TXAWSTestCase


class AWSServiceEndpointTestCase(TXAWSTestCase):

    def setUp(self):
        self.endpoint = AWSServiceEndpoint(uri="http://my.service/da_endpoint")

    def test_simple_creation(self):
        endpoint = AWSServiceEndpoint()
        self.assertEquals(endpoint.scheme, "http")
        self.assertEquals(endpoint.host, "")
        self.assertEquals(endpoint.port, 80)
        self.assertEquals(endpoint.path, "/")
        self.assertEquals(endpoint.method, "GET")

    def test_parse_uri(self):
        self.assertEquals(self.endpoint.scheme, "http")
        self.assertEquals(self.endpoint.host, "my.service")
        self.assertEquals(self.endpoint.port, 80)
        self.assertEquals(self.endpoint.path, "/da_endpoint")

    def test_parse_uri_https_and_custom_port(self):
        endpoint = AWSServiceEndpoint(uri="https://my.service:8080/endpoint")
        self.assertEquals(endpoint.scheme, "https")
        self.assertEquals(endpoint.host, "my.service")
        self.assertEquals(endpoint.port, 8080)
        self.assertEquals(endpoint.path, "/endpoint")

    def test_custom_method(self):
        endpoint = AWSServiceEndpoint(uri="http://service/endpoint",
                                      method="PUT")
        self.assertEquals(endpoint.method, "PUT")

    def test_get_uri(self):
        uri = self.endpoint.get_uri()
        self.assertEquals(uri, "http://my.service/da_endpoint")

    def test_get_uri_custom_port(self):
        uri = "https://my.service:8080/endpoint"
        endpoint = AWSServiceEndpoint(uri=uri)
        new_uri = endpoint.get_uri()
        self.assertEquals(new_uri, uri)

    def test_set_path(self):
        original_path = self.endpoint.path
        self.endpoint.set_path("/newpath")
        self.assertEquals(
            self.endpoint.get_uri(),
            "http://my.service/newpath")


class AWSServiceRegionTestCase(TXAWSTestCase):

    def setUp(self):
        self.creds = AWSCredentials("foo", "bar")
        self.region = AWSServiceRegion(creds=self.creds)

    def test_simple_creation(self):
        self.assertEquals(self.creds, self.region.creds)
        self.assertEquals(self.region._clients, {})
        self.assertEquals(self.region.ec2_endpoint.get_uri(), EC2_ENDPOINT_US)

    def test_creation_with_keys(self):
        region = AWSServiceRegion(access_key="baz", secret_key="quux")
        self.assertEquals(region.creds.access_key, "baz")
        self.assertEquals(region.creds.secret_key, "quux")

    def test_creation_with_keys_and_creds(self):
        region = AWSServiceRegion(self.creds, access_key="baz",
                                  secret_key="quux")
        self.assertEquals(self.creds.access_key, "foo")
        self.assertEquals(self.creds.secret_key, "bar")

    def test_creation_with_uri(self):
        region = AWSServiceRegion(creds=self.creds, uri="http://foo/bar")
        self.assertEquals(region.ec2_endpoint.get_uri(), "http://foo/bar")

    def test_creation_with_uri_and_region(self):
        region = AWSServiceRegion(creds=self.creds, region=REGION_EU,
                                  uri="http://foo/bar")
        self.assertEquals(region.ec2_endpoint.get_uri(), "http://foo/bar")

    def test_creation_with_region_override(self):
        region = AWSServiceRegion(creds=self.creds, region=REGION_EU)
        self.assertEquals(region.ec2_endpoint.get_uri(), EC2_ENDPOINT_EU)

    def test_get_client_with_empty_cache(self):
        key = str(EC2Client) + str(self.creds) + str(self.region.ec2_endpoint)
        original_client = self.region._clients.get(key)
        new_client = self.region.get_client(
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint)
        self.assertEquals(original_client, None)
        self.assertTrue(isinstance(new_client, EC2Client))
        self.assertNotEquals(original_client, new_client)

    def test_get_client_from_cache(self):
        client1 = self.region.get_client(
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint)
        client2 = self.region.get_client(
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint)
        self.assertTrue(isinstance(client1, EC2Client))
        self.assertTrue(isinstance(client2, EC2Client))
        self.assertEquals(client1, client2)

    def test_get_client_from_cache_with_purge(self):
        client1 = self.region.get_client(
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint,
            purge_cache=True)
        client2 = self.region.get_client(
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint,
            purge_cache=True)
        self.assertTrue(isinstance(client1, EC2Client))
        self.assertTrue(isinstance(client2, EC2Client))
        self.assertNotEquals(client1, client2)

    def test_get_ec2_client_from_cache(self):
        client1 = self.region.get_ec2_client(self.creds)
        client2 = self.region.get_ec2_client(self.creds)
        self.assertEquals(self.creds, self.region.creds)
        self.assertTrue(isinstance(client1, EC2Client))
        self.assertTrue(isinstance(client2, EC2Client))
        self.assertEquals(client1, client2)


    def test_get_s3_client(self):
        self.assertRaises(NotImplementedError, self.region.get_s3_client)

    def test_get_simpledb_client(self):
        self.assertRaises(NotImplementedError, self.region.get_simpledb_client)

    def test_get_sqs_client(self):
        self.assertRaises(NotImplementedError, self.region.get_sqs_client)

{'fraction_non_alphanumeric': 0.10881438534576926, 'fraction_numerical': 0.01209982354423998, 'mean_word_length': 3.2751436781609193, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5803162', 'n_tokens_mistral': 3780, 'n_tokens_neox': 3644, 'n_words': 512}