def test_make_router_status_entry_with_live_descriptor(self):
    """
    Tests creation of router status entries with a live server descriptor.
    """

    with open(get_resource('server_descriptor_with_ed25519'), 'rb') as descriptor_file:
      desc = next(stem.descriptor.parse_file(descriptor_file, validate = True)).make_router_status_entry()

    self.assertEqual(stem.descriptor.router_status_entry.RouterStatusEntryV3, type(desc))
    self.assertEqual('destiny', desc.nickname)
    self.assertEqual('F65E0196C94DFFF48AFBF2F5F9E3E19AAE583FD0', desc.fingerprint)
    self.assertEqual(datetime.datetime(2015, 8, 22, 15, 21, 45), desc.published)
    self.assertEqual('94.242.246.23', desc.address)
    self.assertEqual(9001, desc.or_port)
    self.assertEqual(443, desc.dir_port)
    self.assertEqual(['Fast', 'Named', 'Running', 'Stable', 'Valid'], desc.flags)
    self.assertEqual(stem.version.Version('0.2.7.2-alpha-dev'), desc.version)
    self.assertEqual('Tor 0.2.7.2-alpha-dev', desc.version_line)

    self.assertEqual([('2a01:608:ffff:ff07::1:23', 9003, True)], desc.or_addresses)
    self.assertEqual('ed25519', desc.identifier_type)
    self.assertEqual('pbYagEQPUiNjcDp/oY2oESXkDzd8PZlr26kaR7nUkao', desc.identifier)
    self.assertEqual('B5E441051D139CCD84BC765D130B01E44DAC29AD', desc.digest)
    self.assertEqual(149715200, desc.bandwidth)
    self.assertEqual(None, desc.measured)
    self.assertEqual(False, desc.is_unmeasured)
    self.assertEqual([], desc.unrecognized_bandwidth_entries)
    self.assertEqual(stem.exit_policy.MicroExitPolicy('reject 25,465,587,10000,14464'), desc.exit_policy)
    self.assertEqual([], desc.microdescriptor_hashes)

  def test_bridge_descriptor(self):
    """
    Parses a bridge descriptor.
    """

    with open(get_resource('bridge_descriptor'), 'rb') as descriptor_file:
      desc = next(stem.descriptor.parse_file(descriptor_file, 'bridge-server-descriptor 1.0', validate = True))

    self.assertEqual('Unnamed', desc.nickname)
    self.assertEqual('4ED573582B16ACDAF6E42AA044A038F83A7F6333', desc.fingerprint)
    self.assertEqual('10.18.111.71', desc.address)
    self.assertEqual(9001, desc.or_port)
    self.assertEqual(None, desc.socks_port)
    self.assertEqual(None, desc.dir_port)
    self.assertEqual(b'Tor 0.2.0.26-rc (r14597) on Linux i686', desc.platform)
    self.assertEqual(stem.version.Version('0.2.0.26-rc'), desc.tor_version)
    self.assertEqual('Linux i686', desc.operating_system)
    self.assertEqual(204, desc.uptime)
    self.assertEqual(datetime.datetime(2008, 5, 20, 19, 45, 0), desc.published)
    self.assertEqual(None, desc.contact)
    self.assertEqual(['1', '2'], desc.link_protocols)
    self.assertEqual(['1'], desc.circuit_protocols)
    self.assertEqual(False, desc.hibernating)
    self.assertEqual(False, desc.allow_single_hop_exits)
    self.assertEqual(True, desc.extra_info_cache)
    self.assertEqual('BB1F13AA431421BEA29B840A2E33BB1C31C2990B', desc.extra_info_digest)
    self.assertEqual(None, desc.hidden_service_dir)
    self.assertEqual(set(), desc.family)
    self.assertEqual(3220480, desc.average_bandwidth)
    self.assertEqual(6441984, desc.burst_bandwidth)
    self.assertEqual(59408, desc.observed_bandwidth)
    self.assertEqual(stem.exit_policy.ExitPolicy('reject *:*'), desc.exit_policy)
    self.assertEqual('00F1CD29AD308A59A9AB5A88B49ECB46E0F215FD', desc.digest())
    self.assertEqual([], desc.get_unrecognized_lines())

  def test_metrics_bridge_descriptor(self):
    """
    Parses and checks our results against an extrainfo bridge descriptor from
    metrics.
    """

    descriptor_file = open(get_resource('extrainfo_bridge_descriptor'), 'rb')

    expected_dir_v2_responses = {
      DirResponse.OK: 0,
      DirResponse.UNAVAILABLE: 0,
      DirResponse.NOT_FOUND: 0,
      DirResponse.NOT_MODIFIED: 0,
      DirResponse.BUSY: 0,
    }

    expected_dir_v3_responses = {
      DirResponse.OK: 72,
      DirResponse.NOT_ENOUGH_SIGS: 0,
      DirResponse.UNAVAILABLE: 0,
      DirResponse.NOT_FOUND: 0,
      DirResponse.NOT_MODIFIED: 0,
      DirResponse.BUSY: 0,
    }

    desc = next(stem.descriptor.parse_file(descriptor_file, 'bridge-extra-info 1.0'))
    self.assertEqual('ec2bridgereaac65a3', desc.nickname)
    self.assertEqual('1EC248422B57D9C0BD751892FE787585407479A4', desc.fingerprint)
    self.assertEqual(datetime.datetime(2012, 6, 8, 2, 21, 27), desc.published)
    self.assertEqual(datetime.datetime(2012, 6, 8, 2, 10, 38), desc.read_history_end)
    self.assertEqual(900, desc.read_history_interval)
    self.assertEqual(datetime.datetime(2012, 6, 8, 2, 10, 38), desc.write_history_end)
    self.assertEqual(900, desc.write_history_interval)
    self.assertEqual(datetime.datetime(2012, 6, 8, 2, 10, 38), desc.dir_read_history_end)
    self.assertEqual(900, desc.dir_read_history_interval)
    self.assertEqual(datetime.datetime(2012, 6, 8, 2, 10, 38), desc.dir_write_history_end)
    self.assertEqual(900, desc.dir_write_history_interval)
    self.assertEqual('00A2AECCEAD3FEE033CFE29893387143146728EC', desc.digest())
    self.assertEqual([], desc.get_unrecognized_lines())

    read_values_start = [337920, 437248, 3995648, 48726016]
    self.assertEqual(read_values_start, desc.read_history_values[:4])

    write_values_start = [343040, 991232, 5649408, 49548288]
    self.assertEqual(write_values_start, desc.write_history_values[:4])

    dir_read_values_start = [0, 71680, 99328, 25600]
    self.assertEqual(dir_read_values_start, desc.dir_read_history_values[:4])

    dir_write_values_start = [5120, 664576, 2419712, 578560]
    self.assertEqual(dir_write_values_start, desc.dir_write_history_values[:4])

    self.assertEqual({}, desc.dir_v2_requests)
    self.assertEqual({}, desc.dir_v3_requests)

    self.assertEqual(expected_dir_v2_responses, desc.dir_v2_responses)
    self.assertEqual(expected_dir_v3_responses, desc.dir_v3_responses)

    self.assertEqual({}, desc.dir_v2_responses_unknown)
    self.assertEqual({}, desc.dir_v2_responses_unknown)

  def test_for_duckduckgo_without_validation(self):
    """
    Parse duckduckgo's descriptor
    """

    descriptor_file = open(get_resource('hidden_service_duckduckgo'), 'rb')
    desc = next(stem.descriptor.parse_file(descriptor_file, 'hidden-service-descriptor 1.0', validate = False))
    self._assert_matches_duckduckgo(desc)

  def test_with_ed25519_expired_cert(self):
    """
    Parses a server descriptor with an expired ed25519 certificate
    """

    desc_text = open(get_resource('bridge_descriptor_with_ed25519'), 'rb').read()
    desc_iter = stem.descriptor.server_descriptor._parse_file(io.BytesIO(desc_text), validate = True)
    self.assertRaises(ValueError, list, desc_iter)

  def test_nonascii_v3_reqs(self):
    """
    Malformed descriptor with non-ascii content for the 'dirreq-v3-reqs' line.
    """

    with open(get_resource('unparseable/extrainfo_nonascii_v3_reqs'), 'rb') as descriptor_file:
      desc_generator = stem.descriptor.parse_file(descriptor_file, 'extra-info 1.0', validate = True)
      exc_msg = "'dirreq-v3-reqs' line had non-ascii content: S?=4026597208,S?=4026597208,S?=4026597208,S?=4026597208,S?=4026597208,S?=4026597208,??=4026591624,6?=4026537520,6?=4026537520,6?=4026537520,us=8"
      self.assertRaisesRegexp(ValueError, re.escape(exc_msg), next, desc_generator)

  def test_with_tarfile_path(self):
    """
    Fetch server descriptors via parse_file() for a tarfile path.
    """

    descriptors = list(stem.descriptor.parse_file(get_resource('descriptor_archive.tar')))
    self.assertEqual(3, len(descriptors))

    fingerprints = set([desc.fingerprint for desc in descriptors])
    self.assertEqual(TARFILE_FINGERPRINTS, fingerprints)

  def test_validation_with_descriptor_key(self):
    """
    Validate a descriptor signature using the ed25519 master key within the
    descriptor.
    """

    with open(get_resource('server_descriptor_with_ed25519'), 'rb') as descriptor_file:
      desc = next(stem.descriptor.parse_file(descriptor_file, validate = False))

    desc.certificate.validate(desc)

  def test_validation_with_invalid_descriptor(self):
    """
    Validate a descriptor without a valid signature.
    """

    with open(get_resource('server_descriptor_with_ed25519'), 'rb') as descriptor_file:
      desc = next(stem.descriptor.parse_file(descriptor_file, validate = False))

    cert = Ed25519Certificate.parse(certificate())
    self.assertRaisesRegexp(ValueError, re.escape('Ed25519KeyCertificate signing key is invalid (Signature was forged or corrupt)'), cert.validate, desc)

  def test_nonascii_v3_reqs(self):
    """
    Malformed descriptor with non-ascii content for the 'dirreq-v3-reqs' line.
    """

    with open(get_resource('extrainfo_nonascii_v3_reqs'), 'rb') as descriptor_file:
      try:
        next(stem.descriptor.parse_file(descriptor_file, 'extra-info 1.0', validate = True))
        self.fail("validation should've raised an exception")
      except ValueError as exc:
        expected = "'dirreq-v3-reqs' line had non-ascii content: S?=4026597208,S?=4026597208,S?=4026597208,S?=4026597208,S?=4026597208,S?=4026597208,??=4026591624,6?=4026537520,6?=4026537520,6?=4026537520,us=8"
        self.assertEqual(expected, str(exc))

  def test_with_ed25519(self):
    """
    Parses a descriptor with a ed25519 identity key, as added by proposal 228
    (cross certification onionkeys).
    """

    with open(get_resource('server_descriptor_with_ed25519'), 'rb') as descriptor_file:
      desc = next(stem.descriptor.parse_file(descriptor_file, validate = True))

    family = set([
      '$379FB450010D17078B3766C2273303C358C3A442',
      '$3EB46C1D8D8B1C0BBCB6E4F08301EF68B7F5308D',
      '$B0279A521375F3CB2AE210BDBFC645FDD2E1973A',
      '$EC116BCB80565A408CE67F8EC3FE3B0B02C3A065',
    ])

    self.assertEqual('destiny', desc.nickname)
    self.assertEqual('F65E0196C94DFFF48AFBF2F5F9E3E19AAE583FD0', desc.fingerprint)
    self.assertEqual('94.242.246.23', desc.address)
    self.assertEqual(9001, desc.or_port)
    self.assertEqual(None, desc.socks_port)
    self.assertEqual(443, desc.dir_port)
    self.assertTrue('bWPo2fIzo3uOywfoM' in desc.ed25519_certificate)
    self.assertEqual('Z6a1UabSK+N21j6NnyM6N7jssH6DK68qa6W5uB4QpGQ', desc.ed25519_master_key)
    self.assertEqual('w+cKNZTlL7vz/4WgYdFUblzJy3VdTw0mfFK4N3SPFCt20fNKt9SgiZ5V/2ai3kgGsc6oCsyUesSiYtPcTXMLCw', desc.ed25519_signature)
    self.assertEqual(b'Tor 0.2.7.2-alpha-dev on Linux', desc.platform)
    self.assertEqual(stem.version.Version('0.2.7.2-alpha-dev'), desc.tor_version)
    self.assertEqual('Linux', desc.operating_system)
    self.assertEqual(1362680, desc.uptime)
    self.assertEqual(datetime.datetime(2015, 8, 22, 15, 21, 45), desc.published)
    self.assertEqual(b'0x02225522 Frenn vun der Enn (FVDE) <info AT enn DOT lu>', desc.contact)
    self.assertEqual(['1', '2'], desc.link_protocols)
    self.assertEqual(['1'], desc.circuit_protocols)
    self.assertEqual(False, desc.hibernating)
    self.assertEqual(False, desc.allow_single_hop_exits)
    self.assertEqual(False, desc.extra_info_cache)
    self.assertEqual('44E9B679AF0B4EB09296985BAF4066AE9CA5BB93', desc.extra_info_digest)
    self.assertEqual(['2'], desc.hidden_service_dir)
    self.assertEqual(family, desc.family)
    self.assertEqual(149715200, desc.average_bandwidth)
    self.assertEqual(1048576000, desc.burst_bandwidth)
    self.assertEqual(51867731, desc.observed_bandwidth)
    self.assertTrue(desc.exit_policy is not None)
    self.assertEqual(stem.exit_policy.MicroExitPolicy('reject 25,465,587,10000,14464'), desc.exit_policy_v6)
    self.assertTrue('MIGJAoGBAKpPOe' in desc.onion_key)
    self.assertTrue('iW8BqwH5VKqZai' in desc.onion_key_crosscert)
    self.assertTrue('AQoABhtwAWemtV' in desc.ntor_onion_key_crosscert)
    self.assertEqual('0', desc.ntor_onion_key_crosscert_sign)
    self.assertTrue('MIGJAoGBAOUS7x' in desc.signing_key)
    self.assertTrue('y72z1dZOYxVQVL' in desc.signature)
    self.assertEqual('B5E441051D139CCD84BC765D130B01E44DAC29AD', desc.digest())
    self.assertEqual([], desc.get_unrecognized_lines())

  def test_for_facebook(self):
    """
    Parse facebook's descriptor.
    """

    descriptor_file = open(get_resource('hidden_service_facebook'), 'rb')

    desc = next(stem.descriptor.parse_file(descriptor_file, 'hidden-service-descriptor 1.0', validate = True))
    self.assertEqual('utjk4arxqg6s6zzo7n6cjnq6ot34udhr', desc.descriptor_id)
    self.assertEqual(2, desc.version)
    self.assertEqual('6355jaerje3bqozopwq2qmpf4iviizdn', desc.secret_id_part)
    self.assertEqual(datetime.datetime(2014, 10, 31, 23, 0, 0), desc.published)
    self.assertEqual([2, 3], desc.protocol_versions)

  def test_with_ed25519(self):
    """
    Parses a descriptor with a ed25519 identity key.
    """

    with open(get_resource('extrainfo_descriptor_with_ed25519'), 'rb') as descriptor_file:
      desc = next(stem.descriptor.parse_file(descriptor_file, validate = True))

    self.assertEqual('silverfoxden', desc.nickname)
    self.assertEqual('4970B1DC3DBC8D82D7F1E43FF44B28DBF4765A4E', desc.fingerprint)
    self.assertTrue('AQQABhz0AQFcf5tGWLvPvr' in desc.ed25519_certificate)
    self.assertEqual('g6Zg7Er8K7C1etmt7p20INE1ExIvMRPvhwt6sjbLqEK+EtQq8hT+86hQ1xu7cnz6bHee+Zhhmcc4JamV4eiMAw', desc.ed25519_signature)
    self.assertEqual([], desc.get_unrecognized_lines())

  def test_bridge_with_ed25519(self):
    """
    Parses a bridge descriptor with a ed25519 identity key.
    """

    with open(get_resource('bridge_extrainfo_descriptor_with_ed25519'), 'rb') as descriptor_file:
      desc = next(stem.descriptor.parse_file(descriptor_file, validate = True))

    self.assertEqual('Unnamed', desc.nickname)
    self.assertEqual('B8AB331047F1C1637EFE07FB1B94CCC0FE0ABFFA', desc.fingerprint)
    self.assertFalse(hasattr(desc, 'ed25519_certificate'))
    self.assertEqual('VigmhxML9uw8CT1XeGqZ8KLMhKk6AOKnChQt24usBbI', desc.ed25519_certificate_hash)
    self.assertEqual('7DSOQz9eGgjDX6GT7qcrVViK8yqJD4aoEnuhdAgYtgA', desc.router_digest_sha256)
    self.assertEqual([], desc.get_unrecognized_lines())

  def test_with_carriage_returns(self):
    """
    Read a descriptor file with windows newlines (CRLF).
    """

    descriptor_path = get_resource('unparseable/cached-microdesc-consensus_with_carriage_returns')

    with open(descriptor_path, 'rb') as descriptor_file:
      descriptors = stem.descriptor.parse_file(descriptor_file, 'network-status-microdesc-consensus-3 1.0', normalize_newlines = True)

      # if we didn't strip \r then it would be part of the last flag

      router = next(descriptors)
      self.assertEqual([Flag.FAST, Flag.RUNNING, Flag.STABLE, Flag.VALID], router.flags)

  def test_with_tarfile_object(self):
    """
    Fetch server descriptors via parse_file() for a tarfile object.
    """

    # TODO: When dropping python 2.6 support we can go back to using the 'with'
    # keyword here.

    tar_file = tarfile.open(get_resource('descriptor_archive.tar'))
    descriptors = list(stem.descriptor.parse_file(tar_file))
    self.assertEqual(3, len(descriptors))

    fingerprints = set([desc.fingerprint for desc in descriptors])
    self.assertEqual(TARFILE_FINGERPRINTS, fingerprints)
    tar_file.close()

  def test_multiple_metrics_bridge_descriptors(self):
    """
    Check that we can read bridge descriptors when there's multiple in a file.
    """

    descriptor_file = open(get_resource('extrainfo_bridge_descriptor_multiple'), 'rb')
    desc_list = list(stem.descriptor.parse_file(descriptor_file))

    self.assertEqual(6, len(desc_list))
    self.assertEqual('909B07DB17E21D263C55794AB815BF1DB195FDD9', desc_list[0].fingerprint)
    self.assertEqual('7F7798A3CBB0F643B1CFCE3FD4F2B7C553764498', desc_list[1].fingerprint)
    self.assertEqual('B4869206C1EEA4A090FE614155BD6942701F80F1', desc_list[2].fingerprint)
    self.assertEqual('C18896EB6274DC8123491FAE1DD17E1769C54C4F', desc_list[3].fingerprint)
    self.assertEqual('478B4CB438302981DE9AAF246F48DBE57F69050A', desc_list[4].fingerprint)
    self.assertEqual('25D9D52A0350B42E69C8AB7CE945DB1CA38DA0CF', desc_list[5].fingerprint)

  def test_metrics_descriptor_multiple(self):
    """
    Parses and checks our results against a server descriptor from metrics.
    """

    with open(get_resource('metrics_server_desc_multiple'), 'rb') as descriptor_file:
      descriptors = list(stem.descriptor.parse_file(descriptor_file, 'server-descriptor 1.0'))

      self.assertEqual(2, len(descriptors))

      self.assertEqual('anonion', descriptors[0].nickname)
      self.assertEqual('9A5EC5BB866517E53962AF4D3E776536694B069E', descriptors[0].fingerprint)

      self.assertEqual('Unnamed', descriptors[1].nickname)
      self.assertEqual('5366F1D198759F8894EA6E5FF768C667F59AFD24', descriptors[1].fingerprint)

  def test_bridge_with_ed25519(self):
    """
    Parses a bridge descriptor with ed25519.
    """

    with open(get_resource('bridge_descriptor_with_ed25519'), 'rb') as descriptor_file:
      desc = next(stem.descriptor.parse_file(descriptor_file, validate = True))

    self.assertEqual('ChandlerObfs11', desc.nickname)
    self.assertEqual('678912ABD7398DF8EFC8FA2BC7DEF610710360C4', desc.fingerprint)
    self.assertEqual('10.162.85.172', desc.address)
    self.assertFalse(hasattr(desc, 'ed25519_certificate'))
    self.assertEqual('lgIuiAJCoXPRwWoHgG4ZAoKtmrv47aPr4AsbmESj8AA', desc.ed25519_certificate_hash)
    self.assertEqual('OB/fqLD8lYmjti09R+xXH/D4S2qlizxdZqtudnsunxE', desc.router_digest_sha256)
    self.assertEqual([], desc.get_unrecognized_lines())

  def test_with_basic_auth(self):
    """
    Parse a descriptor with introduction-points encrypted with basic auth.
    """

    if not stem.prereq.is_crypto_available():
      return test.runner.skip(self, 'requires pycrypto')

    descriptor_file = open(get_resource('hidden_service_basic_auth'), 'rb')

    desc = next(stem.descriptor.parse_file(descriptor_file, 'hidden-service-descriptor 1.0', validate = True))
    self.assertEqual('yfmvdrkdbyquyqk5vygyeylgj2qmrvrd', desc.descriptor_id)
    self.assertEqual(2, desc.version)
    self.assertEqual('fluw7z3s5cghuuirq3imh5jjj5ljips6', desc.secret_id_part)
    self.assertEqual(datetime.datetime(2015, 2, 24, 20, 0, 0), desc.published)
    self.assertEqual([2, 3], desc.protocol_versions)
    self.assertEqual(EXPECTED_BASIC_AUTH_INTRODUCTION_POINTS_ENCODED, desc.introduction_points_encoded)
    self.assertEqual([], desc.introduction_points_auth)

    self.assertRaises(DecryptionFailure, desc.introduction_points)
    self.assertRaises(DecryptionFailure, desc.introduction_points, 'aCmx3qIvArbil8A0KM4KgQ==')

    introduction_points = desc.introduction_points('dCmx3qIvArbil8A0KM4KgQ==')
    self.assertEqual(3, len(introduction_points))

    point = introduction_points[0]
    self.assertEqual('hmtvoobwglmmec26alnvl7x7mgmmr7xv', point.identifier)
    self.assertEqual('195.154.82.88', point.address)
    self.assertEqual(443, point.port)
    self.assertTrue('MIGJAoGBANbPRD07T' in point.onion_key)
    self.assertTrue('MIGJAoGBAN+LAdZP/' in point.service_key)
    self.assertEqual([], point.intro_authentication)

    point = introduction_points[1]
    self.assertEqual('q5w6l2f4g5zw4rkr56fkyovbkkrnzcj5', point.identifier)
    self.assertEqual('37.252.190.133', point.address)
    self.assertEqual(9001, point.port)
    self.assertTrue('MIGJAoGBAKmsbKrtt' in point.onion_key)
    self.assertTrue('MIGJAoGBANwczLtzR' in point.service_key)
    self.assertEqual([], point.intro_authentication)

    point = introduction_points[2]
    self.assertEqual('qcvprvmvnjb4dfyqjtxskugniliwlrx3', point.identifier)
    self.assertEqual('193.11.114.45', point.address)
    self.assertEqual(9002, point.port)
    self.assertTrue('MIGJAoGBAM1ILL+7P' in point.onion_key)
    self.assertTrue('MIGJAoGBAM7B/cymp' in point.service_key)
    self.assertEqual([], point.intro_authentication)