def test_decoding(self):
   self.assertEqual(rfc1924_b85decode(mercurial_encoded), mercurial_bytes)
   self.assertEqual(rfc1924_b85decode(random_256_mercurial),
                    random_256_bytes)
   self.assertEqual(rfc1924_b85decode(b('|NsC0')), b('\xff\xff\xff\xff'))
   for a, e in zip(random_bytes_list, rfc_encoded_bytes_list):
     self.assertEqual(rfc1924_b85decode(e), a)

 def test_query_params_sorted_order(self):
     self.assertEqual(
         generate_base_string_query(dict(b=[8, 2, 4], a=1), {}),
         b("a=1&b=2&b=4&b=8"))
     qs = generate_base_string_query(
         dict(a=5, b=6, c=["w", "a", "t", "e", "r"]), {})
     self.assertEqual(qs, b("a=5&b=6&c=a&c=e&c=r&c=t&c=w"))

def data_urlparse(data_url):
  """
  Parses a data URL into raw bytes and metadata.

  :param data_url:
      The data url string.
      If a mime-type definition is missing in the metadata,
      "text/plain;charset=US-ASCII" will be used as default mime-type.
  :returns:
      A 2-tuple::
          (bytes, mime_type)

      See :func:`mom.http.mimeparse.parse_mime_type` for what ``mime_type``
      looks like.
  """
  if not is_bytes(data_url):
    raise TypeError(
      "data URLs must be ASCII-encoded bytes: got %r" %
      type(data_url).__name__
    )
  metadata, encoded = data_url.rsplit(b(","), 1)
  _, metadata = metadata.split(b("data:"), 1)
  parts = metadata.rsplit(b(";"), 1)
  if parts[-1] == b("base64"):
    decode = base64_decode
    parts = parts[:-1]
  else:
    decode = unquote
  if not parts or not parts[0]:
    parts = [b("text/plain;charset=US-ASCII")]
  mime_type = parse_mime_type(parts[0])
  raw_bytes = decode(encoded)
  return raw_bytes, mime_type

def best_match(supported, header):
    """Return mime-type with the highest quality ('q') from list of candidates.

    Takes a list of supported mime-types and finds the best match for all the
    media-ranges listed in header. The value of header must be a string that
    conforms to the format of the HTTP Accept: header. The value of 'supported'
    is a list of mime-types. The list of supported mime-types should be sorted
    in order of increasing desirability, in case of a situation where there is
    a tie.

    >>> best_match([b'application/xbel+xml', b'text/xml'],
                   b'text/*;q=0.5,*/*; q=0.1')
    b'text/xml'
    """
    split_header = _filter_blank(header.split(b(',')))
    parsed_header = [parse_media_range(r) for r in split_header]
    weighted_matches = []
    pos = 0
    for mime_type in supported:
        weighted_matches.append((fitness_and_quality_parsed(mime_type,
                                 parsed_header), pos, mime_type))
        pos += 1
    weighted_matches.sort()

    return weighted_matches[-1][0][1] and weighted_matches[-1][2] or b('')

 def test_parsing_no_metadata(self):
   raw_bytes, mime_type = data_urlparse(no_meta_data_url)
   self.assertEqual(raw_bytes, png)
   self.assertEqual(mime_type[:2], (b('text'), b('plain')))
   self.assertDictEqual(mime_type[2], {
     b('charset'): b('US-ASCII'),
     })

 def test_overflow_allowed(self):
   self.assertEqual(uint_to_bytes(0xc0ff, fill_size=1, overflow=True),
                    b('\xc0\xff'))
   self.assertEqual(uint_to_bytes(123456789, fill_size=3, overflow=True),
                    b('\x07\x5b\xcd\x15'))
   self.assertEqual(uint_to_bytes(0xf00dc0ffee, fill_size=4,
                                  overflow=True),
                    b('\xf0\x0d\xc0\xff\xee'))

 def test_when_token_secret_present(self):
     base_string = generate_base_string(
         HTTP_POST, b("http://example.com/"), self.oauth_params)
     self.assertEqual(generate_plaintext_signature(
         base_string,
         b(""),
         self.oauth_token_secret
     ), b("&token%20test%20secret"))

 def test_when_neither_secret_present(self):
     base_string = generate_base_string(
         HTTP_POST, b("http://example.com/"), self.oauth_params)
     self.assertEqual(generate_plaintext_signature(
         base_string,
         b(""),
         None
     ), b("&"))

    def test_raises_InvalidHttpRequestError_when_identifier_invalid(self):
        temporary_credentials = Credentials(identifier=RFC_TEMPORARY_IDENTIFIER,
                                            shared_secret=RFC_TEMPORARY_SECRET)

        self.assertRaises(InvalidHttpRequestError,
                          _OAuthClient.check_verification_code,
                          temporary_credentials, b("non-matching-token"),
                          b("verification-code"))

 def test_fill_size(self):
   self.assertEqual(uint_to_bytes(0xc0ff, fill_size=4),
                    b('\x00\x00\xc0\xff'))
   self.assertEqual(uint_to_bytes(0xc0ffee, fill_size=6),
                    b('\x00\x00\x00\xc0\xff\xee'))
   self.assertEqual(uint_to_bytes(123456789, fill_size=6),
                    b('\x00\x00\x07[\xcd\x15'))
   self.assertEqual(uint_to_bytes(123456789, fill_size=7),
                    b('\x00\x00\x00\x07[\xcd\x15'))

 def setUp(self):
     self.oauth_params = dict(
         oauth_consumer_key=b("9djdj82h48djs9d2"),
         oauth_token=b("kkk9d7dh3k39sjv7"),
         oauth_signature_method=SIGNATURE_METHOD_HMAC_SHA1,
         oauth_timestamp=b("137131201"),
         oauth_nonce=b("7d8f3e4a"),
         oauth_signature=b("bYT5CMsGcbgUdFHObYMEfcx6bsw%3D")
     )

 def test_raises_InvalidHttpRequestError_when_body_and_GET(self):
     oauth_params = dict(
         oauth_blah=b("blah"),
     )
     self.assertRaises(InvalidHttpRequestError,
                       _OAuthClient._build_request,
                       HTTP_GET,
                       FOO_URI,
                       None, b("a=b"), {}, oauth_params,
                       OAUTH_REALM, False)

 def test_codec_equivalence(self):
   # Padding bytes are not preserved (it is acceptable here).
   random_bytes = b("\x00\xbcE\x9a\xda]")
   expected_bytes = b("\xbcE\x9a\xda]")
   self.assertEqual(uint_to_bytes(bytes_to_uint(random_bytes)),
                    expected_bytes)
   self.assertEqual(uint_to_bytes(bytes_to_uint_naive(random_bytes)),
                    expected_bytes)
   self.assertEqual(uint_to_bytes(bytes_to_uint_simple(random_bytes)),
                    expected_bytes)

  def test_ValueError_when_encoded_length_not_20(self):
    self.assertRaises(ValueError, ipv6_b85decode,
                      b('=r54lj&NUUO~Hi%c2ym0='))
    self.assertRaises(ValueError, ipv6_b85decode,
                      b('=r54lj&NUUO='))

    self.assertRaises(ValueError, ipv6_b85decode_naive,
                      b('=r54lj&NUUO~Hi%c2ym0='))
    self.assertRaises(ValueError, ipv6_b85decode_naive,
                      b('=r54lj&NUUO='))

 def test_get_authentication_url(self):
     url = self.client.get_authentication_url(self.temporary_credentials,
                                             a="something here",
                                             b=["another thing", 5],
                                             oauth_ignored=b("ignored"))
     self.assertEqual(url,
                      RFC_AUTHENTICATION_URI +
                      b("?a=something%20here"
                      "&b=5"
                      "&b=another%20thing&oauth_token=") +
                      self.temporary_credentials.identifier)

    def test__resource_request_data(self):
        expected = RequestAdapter(
            HTTP_GET,
            RFC_RESOURCE_FULL_URL,
            b(''),
            headers = {
                HEADER_AUTHORIZATION_CAPS: b('''\
OAuth realm="Photos",\
oauth_consumer_key="dpf43f3p2l4k3l03",\
oauth_token="nnch734d00sl2jdk",\
oauth_signature_method="HMAC-SHA1",\
oauth_timestamp="137131202",\
oauth_nonce="chapoH",\
oauth_signature="MdpQcU8iPSUjWoN%2FUDMsK2sui9I%3D"'''),
            }
        )
        auth_credentials = Credentials(RFC_TOKEN_IDENTIFIER,
                                       RFC_TOKEN_SECRET)
        client_credentials = Credentials(RFC_CLIENT_IDENTIFIER,
                                         RFC_CLIENT_SECRET)

        class MockClient(_OAuthClient):
            @classmethod
            def generate_timestamp(cls):
                return RFC_TIMESTAMP_3

            @classmethod
            def generate_nonce(cls):
                return RFC_NONCE_3

        got = MockClient._request(
            client_credentials,
            HTTP_GET,
            RFC_RESOURCE_URI,
            params={
                "file": b("vacation.jpg"),
                "size": b("original"),
            },
            realm=RFC_REALM,
            auth_credentials=auth_credentials,
            oauth_version=None
        )
        self.assertEqual(got.method, expected.method)
        self.assertEqual(got.url, expected.url)
        self.assertEqual(got.body, expected.body)
        expected_headers, expected_realm = parse_authorization_header(
            expected.headers[HEADER_AUTHORIZATION_CAPS],
        )
        got_headers, got_realm = parse_authorization_header(
            got.headers[HEADER_AUTHORIZATION_CAPS],
        )
        self.assertDictEqual(got_headers, expected_headers)
        self.assertEqual(got_realm, expected_realm)

 def test_base_string_does_not_contain_oauth_signature(self):
     # Ensure both are present in the query parameters as well as the URL.
     oauth_params = {
         OAUTH_PARAM_REALM: b("example.com"),
     }
     oauth_params.update(self.oauth_params)
     url = b("http://example.com/request?"
           "oauth_signature=foobar&realm=something")
     base_string = generate_base_string(HTTP_POST, url, oauth_params)
     self.assertTrue(b("oauth_signature%3D") not in base_string)
     self.assertTrue(b("realm%3Dexample.com") not in base_string)
     self.assertTrue(b("realm%3Dsomething") in base_string)

 def test_raises_InvalidAuthorizationHeaderError_when_auth_present(self):
     oauth_params = dict(
         oauth_blah=b("blah"),
     )
     self.assertRaises(InvalidAuthorizationHeaderError,
                       _OAuthClient._build_request,
                       HTTP_POST,
                       FOO_URI,
                       None, b("a=b"),
             {
                 HEADER_AUTHORIZATION_CAPS: b("")
         }, oauth_params,
                       OAUTH_REALM, False)

  def test_zero(self):
    self.assertEqual(uint_to_bytes(0), b('\x00'))
    self.assertEqual(uint_to_bytes(0, 4), b('\x00') * 4)
    self.assertEqual(uint_to_bytes(0, 7), b('\x00') * 7)
    self.assertEqual(uint_to_bytes(0, chunk_size=1), b('\x00'))
    self.assertEqual(uint_to_bytes(0, chunk_size=4), b('\x00') * 4)
    self.assertEqual(uint_to_bytes(0, chunk_size=7), b('\x00') * 7)

    self.assertEqual(uint_to_bytes_naive(0), b('\x00'))
    self.assertEqual(uint_to_bytes_naive(0, 4), b('\x00') * 4)
    self.assertEqual(uint_to_bytes_naive(0, 7), b('\x00') * 7)

    self.assertEqual(uint_to_bytes_simple(0), b('\x00'))

    def test_oauth_signature_and_realm_are_excluded_properly(self):
        qs = generate_base_string_query({
            OAUTH_PARAM_SIGNATURE: "something"
            },
            self.specification_example_oauth_params
        )
        self.assertTrue(b("oauth_signature=") not in qs)
        self.assertTrue(b("realm=") not in qs)

        self.assertTrue(
            generate_base_string_query(dict(realm="something"), dict()),
            b("realm=something")
        )