def test_multientry_keywords_2(self):
     x = parse_keywords("foo=bar\nfoo=zarimba")
     self.assertEqual(len(x), 1)
     self.assertTrue(isinstance(x["foo"], types.ListType))
     self.assertEqual(len(x["foo"]), 2)
     self.assertEqual(x["foo"][0], "bar")
     self.assertEqual(x["foo"][1], "zarimba")

    def _conf_changed(self, arg):
        """
        internal callback. from control-spec:

        4.1.18. Configuration changed

          The syntax is:
             StartReplyLine *(MidReplyLine) EndReplyLine

             StartReplyLine = "650-CONF_CHANGED" CRLF
             MidReplyLine = "650-" KEYWORD ["=" VALUE] CRLF
             EndReplyLine = "650 OK"

          Tor configuration options have changed (such as via a SETCONF or
          RELOAD signal). KEYWORD and VALUE specify the configuration option
          that was changed.  Undefined configuration options contain only the
          KEYWORD.
        """

        conf = parse_keywords(arg, multiline_values=False)
        for (k, v) in conf.items():
            # v will be txtorcon.DEFAULT_VALUE already from
            # parse_keywords if it was unspecified
            real_name = self._find_real_name(k)
            if real_name in self.parsers:
                v = self.parsers[real_name].parse(v)
            self.config[real_name] = v

 def test_multientry_keywords_3(self):
     x = parse_keywords('foo=bar\nfoo=baz\nfoo=zarimba')
     self.assertEqual(len(x), 1)
     self.assertTrue(isinstance(x['foo'], list))
     self.assertEqual(len(x['foo']), 3)
     self.assertEqual(x['foo'][0], 'bar')
     self.assertEqual(x['foo'][1], 'baz')
     self.assertEqual(x['foo'][2], 'zarimba')

    def test_multientry_keywords_2(self):
        x = parse_keywords('''foo=bar
foo=zarimba''')
        self.assertTrue(len(x) == 1)
        self.assertTrue(isinstance(x['foo'], types.ListType))
        self.assertTrue(len(x['foo']) == 2)
        self.assertTrue(x['foo'][0] == 'bar')
        self.assertTrue(x['foo'][1] == 'zarimba')

 def test_keywords(self):
     x = parse_keywords(
         "events/names=CIRC STREAM ORCONN BW DEBUG INFO NOTICE WARN ERR NEWDESC ADDRMAP AUTHDIR_NEWDESCS DESCCHANGED NS STATUS_GENERAL STATUS_CLIENT STATUS_SERVER GUARD STREAM_BW CLIENTS_SEEN NEWCONSENSUS BUILDTIMEOUT_SET\nOK"
     )
     self.assertTrue("events/names" in x)
     self.assertEqual(
         x["events/names"],
         "CIRC STREAM ORCONN BW DEBUG INFO NOTICE WARN ERR NEWDESC ADDRMAP AUTHDIR_NEWDESCS DESCCHANGED NS STATUS_GENERAL STATUS_CLIENT STATUS_SERVER GUARD STREAM_BW CLIENTS_SEEN NEWCONSENSUS BUILDTIMEOUT_SET",
     )
     self.assertEqual(len(x.keys()), 1)

    def test_multientry_keywords_4(self):
        x = parse_keywords('''foo=bar
foo=baz
foo=zarimba
foo=foo''')
        self.assertEqual(len(x), 1)
        self.assertTrue(isinstance(x['foo'], types.ListType))
        self.assertEqual(len(x['foo']), 4)
        self.assertEqual(x['foo'][0], 'bar')
        self.assertEqual(x['foo'][1], 'baz')
        self.assertEqual(x['foo'][2], 'zarimba')
        self.assertEqual(x['foo'][3], 'foo')

    def test_multiline_keywords_with_spaces(self):
        x = parse_keywords('''ns/name/foo=
r foo aaaam7E7h1vY5Prk8v9/nSRCydY BBBBOfum4CtAYuOgf/D33Qq5+rk 2013-10-27 06:22:18 1.2.3.4 9001 9030
s Fast Guard HSDir Running Stable V2Dir Valid
w Bandwidth=1234
ns/name/bar=
r bar aaaaHgNYtTVPw5hHTO28J4je5i8 BBBBBUaJaBFSU/HDrTxnSh+D3+fY 2013-10-27 07:48:56 1.2.4.5 9001 9030
s Exit Fast Guard HSDir Named Running Stable V2Dir Valid
w Bandwidth=1234
OK
''')
        self.assertEqual(2, len(x))
        keys = sorted(x.keys())
        self.assertEqual(keys, ['ns/name/bar', 'ns/name/foo'])

 def test_default_keywords(self):
     x = parse_keywords('foo')
     self.assertEqual(len(x), 1)
     self.assertTrue(x.has_key('foo'))
     self.assertEqual(x['foo'], DEFAULT_VALUE)

 def test_keywords_mutli_equals(self):
     x = parse_keywords('foo=something subvalue="foo"')
     self.assertTrue(len(x) == 1)
     self.assertTrue(x.has_key('foo'))
     self.assertTrue(x['foo'] == 'something subvalue="foo"')

    def _bootstrap(self, arg=None):
        "This takes an arg so we can use it as a callback (see __init__)."

        ## update list of routers (must be before we do the
        ## circuit-status) note that we're feeding each line
        ## incrementally to a state-machine called
        ## _network_status_parser, set up in constructor. "ns" should
        ## be the empty string, but we call _update_network_status for
        ## the de-duplication of named routers

        ns = yield self.protocol.get_info_incremental('ns/all',
                                                      self._network_status_parser.process)
        self._update_network_status(ns)

        ## update list of existing circuits
        cs = yield self.protocol.get_info_raw('circuit-status')
        self._circuit_status(cs)

        ## update list of streams
        ss = yield self.protocol.get_info_raw('stream-status')
        self._stream_status(ss)

        ## update list of existing address-maps
        key = 'address-mappings/all'
        am = yield self.protocol.get_info_raw(key)
        ## strip addressmappsings/all= and OK\n from raw data
        am = am[len(key) + 1:]
        if am.strip() != 'OK':
            for line in am.split('\n')[:-1]:
                if len(line.strip()) == 0:
                    continue            # FIXME
                self.addrmap.update(line)

        self._add_events()

        entries = yield self.protocol.get_info_raw("entry-guards")
        for line in entries.split('\n')[1:]:
            if len(line.strip()) == 0 or line.strip() == 'OK':
                continue
            args = line.split()
            (name, status) = args[:2]
            name = name[:41]

            ## this is sometimes redundant, as a missing entry guard
            ## usually means it won't be in our list of routers right
            ## now, but just being on the safe side
            if status.lower() != 'up':
                self.unusable_entry_guards.append(line)
                continue

            try:
                self.entry_guards[name] = self.router_from_id(name)
            except KeyError:
                self.unusable_entry_guards.append(line)

        ## in case process/pid doesn't exist and we don't know the PID
        ## because we own it, we just leave it as 0 (previously
        ## guessed using psutil, but that only works if there's
        ## exactly one tor running anyway)
        try:
            pid = yield self.protocol.get_info_raw("process/pid")
        except TorProtocolError:
            pid = None
        self.tor_pid = 0
        if pid:
            try:
                pid = parse_keywords(pid)['process/pid']
                self.tor_pid = int(pid)
            except KeyError:
                self.tor_pid = 0
        elif self.protocol.is_owned:
            self.tor_pid = self.protocol.is_owned

        self.post_bootstrap.callback(self)
        self.post_boostrap = None

 def test_default_keywords(self):
     x = parse_keywords("foo")
     self.assertEqual(len(x), 1)
     self.assertTrue("foo" in x)
     self.assertEqual(x["foo"], DEFAULT_VALUE)

 def test_unquoted_keywords_empty(self):
     x = parse_keywords("foo=")
     self.assertEqual(x, {"foo": ""})

 def test_unquoted_keywords_singlequote(self):
     x = parse_keywords("Tor='0.1.2.3.4-rc44'")
     self.assertEqual(x, {'Tor': '0.1.2.3.4-rc44'})

    def _bootstrap(self, arg=None):
        "This takes an arg so we can use it as a callback (see __init__)."

        # update list of routers (must be before we do the
        # circuit-status)

        # look out! we're depending on get_info_incremental returning
        # *lines*, which isn't documented -- but will be true because
        # TorControlProtocol is a LineReceiver...
        yield self.protocol.get_info_incremental(
            'ns/all',
            self._network_status_parser.feed_line,
        )
        self._network_status_parser.done()

        # update list of existing circuits
        cs = yield self.protocol.get_info_raw('circuit-status')
        self._circuit_status(cs)

        # update list of streams
        ss = yield self.protocol.get_info_raw('stream-status')
        self._stream_status(ss)

        # update list of existing address-maps
        key = 'address-mappings/all'
        am = yield self.protocol.get_info_raw(key)
        # strip addressmappsings/all= and OK\n from raw data
        am = am[len(key) + 1:]
        for line in am.split('\n'):
            if len(line.strip()) == 0:
                continue            # FIXME
            self.addrmap.update(line)

        yield self._add_events()

        entries = yield self.protocol.get_info_raw("entry-guards")
        for line in entries.split('\n')[1:]:
            if len(line.strip()) == 0 or line.strip() == 'OK':
                # XXX does this ever really happen?
                continue
            args = line.split()
            (name, status) = args[:2]
            name = name[:41]

            # this is sometimes redundant, as a missing entry guard
            # usually means it won't be in our list of routers right
            # now, but just being on the safe side
            if status.lower() != 'up':
                self.unusable_entry_guards.append(line)
                continue

            try:
                self.entry_guards[name] = self.router_from_id(name)
            except KeyError:
                self.unusable_entry_guards.append(line)

        # in case process/pid doesn't exist and we don't know the PID
        # because we own it, we just leave it as 0 (previously
        # guessed using psutil, but that only works if there's
        # exactly one tor running anyway)
        try:
            pid = yield self.protocol.get_info_raw("process/pid")
        except TorProtocolError:
            pid = None
        self.tor_pid = 0
        if pid:
            try:
                pid = parse_keywords(pid)['process/pid']
                self.tor_pid = int(pid)
            except Exception:  # fixme: ValueError and KeyError ..?
                self.tor_pid = 0
        if not self.tor_pid and self.protocol.is_owned:
            self.tor_pid = self.protocol.is_owned

        self.post_bootstrap.callback(self)
        self.post_boostrap = None

    def test_multiline_keywords(self):
        x = parse_keywords('''foo=bar
baz''')
        self.assertTrue(len(x) == 1)
        self.assertTrue(x.has_key('foo'))
        self.assertTrue(x['foo'] == 'bar\nbaz')

 def test_default_keywords(self):
     x = parse_keywords('foo')
     self.assertTrue(len(x) == 1)
     self.assertTrue(x.has_key('foo'))
     self.assertTrue(x['foo'] == DEFAULT_VALUE)

    def test_keywords(self):
        x = parse_keywords("""events/names=CIRC STREAM ORCONN BW DEBUG INFO NOTICE WARN ERR NEWDESC ADDRMAP AUTHDIR_NEWDESCS DESCCHANGED NS STATUS_GENERAL STATUS_CLIENT STATUS_SERVER GUARD STREAM_BW CLIENTS_SEEN NEWCONSENSUS BUILDTIMEOUT_SET
OK""")
        self.assertTrue(x.has_key("events/names"))
        self.assertTrue(x['events/names'] == 'CIRC STREAM ORCONN BW DEBUG INFO NOTICE WARN ERR NEWDESC ADDRMAP AUTHDIR_NEWDESCS DESCCHANGED NS STATUS_GENERAL STATUS_CLIENT STATUS_SERVER GUARD STREAM_BW CLIENTS_SEEN NEWCONSENSUS BUILDTIMEOUT_SET')
        self.assertTrue(len(x.keys()) == 1)

 def test_multiline_keywords(self):
     x = parse_keywords('''Foo=bar\nBar''')
     self.assertEqual(x, {'Foo': 'bar\nBar'})
     x = parse_keywords('''Foo=bar\nBar''', multiline_values=False)
     self.assertEqual(x, {'Foo': 'bar',
                          'Bar': DEFAULT_VALUE})

 def test_unquoted_keywords(self):
     x = parse_keywords('''Tor="0.1.2.3.4-rc44"''')
     self.assertEqual(x, {'Tor': '0.1.2.3.4-rc44'})

 def test_keywords_mutli_equals(self):
     x = parse_keywords('foo=something subvalue="foo"')
     self.assertEqual(len(x), 1)
     self.assertTrue('foo' in x)
     self.assertEqual(x['foo'], 'something subvalue="foo"')