optmanager: cope with bound methods in .subscribe

Fixes #2122

mitmproxy/optmanager.py

@@ -1,9 +1,9 @@
 import contextlib
 import blinker
+import blinker._saferef
 import pprint
 import copy
 import functools
-import weakref
 import os
 import typing
 import textwrap
@@ -127,15 +127,24 @@ class OptManager:
             Subscribe a callable to the .changed signal, but only for a
             specified list of options. The callable should accept arguments
             (options, updated), and may raise an OptionsError.
+
+            The event will automatically be unsubscribed if the callable goes out of scope.
         """
-        func = weakref.proxy(func)
+        for i in opts:
+            if i not in self._options:
+                raise exceptions.OptionsError("No such option: %s" % i)
+
+        # We reuse blinker's safe reference functionality to cope with weakrefs
+        # to bound methods.
+        func = blinker._saferef.safe_ref(func)
 
         @functools.wraps(func)
         def _call(options, updated):
             if updated.intersection(set(opts)):
-                try:
-                    func(options, updated)
-                except ReferenceError:
+                f = func()
+                if f:
+                    f(options, updated)
+                else:
                     self.changed.disconnect(_call)
 
         # Our wrapper function goes out of scope immediately, so we have to set

mitmproxy/tools/console/master.py

@@ -276,7 +276,7 @@ class ConsoleMaster(master.Master):
         self.set_palette(self.options, None)
         self.options.subscribe(
             self.set_palette,
-            ["palette", "palette_transparent"]
+            ["console_palette", "console_palette_transparent"]
         )
         self.loop = urwid.MainLoop(
             urwid.SolidFill("x"),

test/mitmproxy/test_optmanager.py

@@ -151,6 +151,21 @@ def test_subscribe():
     o.two = 4
     assert len(o.changed.receivers) == 0
 
+    class binder:
+        def __init__(self):
+            self.o = TO()
+            self.called = False
+            self.o.subscribe(self.bound, ["two"])
+
+        def bound(self, *args, **kwargs):
+            self.called = True
+
+    t = binder()
+    t.o.one = 3
+    assert not t.called
+    t.o.two = 3
+    assert t.called
+
 
 def test_rollback():
     o = TO()