<commit_before>from __future__ import with_statement

import urllib2
import unittest
import datetime
from nose.tools import eq_

import simples3
from simples3.utils import aws_md5, aws_urlquote, rfc822_fmt
from tests import MockHTTPResponse, BytesIO, g

from tests import setup_package, teardown_package
setup_package, teardown_package

class S3BucketTestCase(unittest.TestCase):
    def setUp(self):
        g.bucket.mock_reset()

    def tearDown(self):
        if g.bucket.mock_responses:
            raise RuntimeError("test run without exhausting mock_responses")

class MiscTests(S3BucketTestCase):
    def test_str(self):
        eq_(str(g.bucket), "<MockBucket johnsmith at "
                           "'http://johnsmith.s3.amazonaws.com'>")

    def test_repr(self):
        eq_(repr(g.bucket),
            "MockBucket('johnsmith', "
            "access_key='<KEY>', "
            "base_url='http://johnsmith.s3.amazonaws.com')")

    def test_timeout_disabled(self):
        g.bucket.timeout = 10.0
        with g.bucket.timeout_disabled():
            eq_(g.bucket.timeout, None)
        eq_(g.bucket.timeout, 10.0)

    def test_error_in_error(self):
        # a hairy situation: an error arising during the parsing of an error.
        def read(bs=4096):
            raise urllib2.URLError("something something dark side")
        FP = type("ErringFP", (object,),
                  {"read": read, "readline": read, "readlines": read})
        url = g.bucket.base_url + "/foo.txt"
        resp = MockHTTPResponse(FP(), {}, url, code=401)
        g.bucket.add_resp_obj(resp, status="401 Something something")
        try:
            g.bucket.get("foo.txt")
        except simples3.S3Error, e:
            assert "read_error" in e.extra

    def test_aws_md5_lit(self):
        val = "Hello!".encode("ascii")
        eq_(aws_md5(val), 'lS0sVtBIWVgzZ0e83ZhZDQ==')

    def test_aws_md5_fp(self):
        val = "Hello world!".encode("ascii")
        eq_(aws_md5(BytesIO(val)), 'hvsmnRkNLIX24EaM7KQqIA==')

    def test_aws_urlquote_funky(self):
        if hasattr(str, "decode"):
            val = "/bucket/\xc3\xa5der".decode("utf-8")
        else:
            val = "/bucket/\xe5der"
        eq_(aws_urlquote(val), "/bucket/%C3%A5der")
<commit_msg>Add test to ensure namespace doesn't change<commit_after>from __future__ import with_statement

import urllib2
import unittest
import datetime
from nose.tools import eq_

import simples3
from simples3.utils import aws_md5, aws_urlquote, rfc822_fmt
from tests import MockHTTPResponse, BytesIO, g

from tests import setup_package, teardown_package
setup_package, teardown_package

class S3BucketTestCase(unittest.TestCase):
    def setUp(self):
        g.bucket.mock_reset()

    def tearDown(self):
        if g.bucket.mock_responses:
            raise RuntimeError("test run without exhausting mock_responses")

class MiscTests(S3BucketTestCase):
    def test_str(self):
        eq_(str(g.bucket), "<MockBucket johnsmith at "
                           "'http://johnsmith.s3.amazonaws.com'>")

    def test_repr(self):
        eq_(repr(g.bucket),
            "MockBucket('johnsmith', "
            "access_key='<KEY>', "
            "base_url='http://johnsmith.s3.amazonaws.com')")

    def test_timeout_disabled(self):
        g.bucket.timeout = 10.0
        with g.bucket.timeout_disabled():
            eq_(g.bucket.timeout, None)
        eq_(g.bucket.timeout, 10.0)

    def test_error_in_error(self):
        # a hairy situation: an error arising during the parsing of an error.
        def read(bs=4096):
            raise urllib2.URLError("something something dark side")
        FP = type("ErringFP", (object,),
                  {"read": read, "readline": read, "readlines": read})
        url = g.bucket.base_url + "/foo.txt"
        resp = MockHTTPResponse(FP(), {}, url, code=401)
        g.bucket.add_resp_obj(resp, status="401 Something something")
        try:
            g.bucket.get("foo.txt")
        except simples3.S3Error, e:
            assert "read_error" in e.extra

    def test_aws_md5_lit(self):
        val = "Hello!".encode("ascii")
        eq_(aws_md5(val), 'lS0sVtBIWVgzZ0e83ZhZDQ==')

    def test_aws_md5_fp(self):
        val = "Hello world!".encode("ascii")
        eq_(aws_md5(BytesIO(val)), 'hvsmnRkNLIX24EaM7KQqIA==')

    def test_aws_urlquote_funky(self):
        if hasattr(str, "decode"):
            val = "/bucket/\xc3\xa5der".decode("utf-8")
        else:
            val = "/bucket/\xe5der"
        eq_(aws_urlquote(val), "/bucket/%C3%A5der")

    def test_amaon_s3_ns_url(self):
      """The amazon namespace needs to be *exactly* as advertised"""
      eq_('http://s3.amazonaws.com/doc/2006-03-01', simples3.bucket.amazon_s3_ns_url)



{'fraction_non_alphanumeric': 0.12352445193929174, 'fraction_numerical': 0.022344013490725127, 'mean_word_length': 2.7509881422924902, 'pattern_counts': {'":': 6, '<': 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': '3908675', 'n_tokens_mistral': 1627, 'n_tokens_neox': 1563, 'n_words': 319}